[ECO-4687][Protocol-2] Feature/resume recover

README.md

@@ -7,7 +7,7 @@
 
 _[Ably](https://ably.com) is the platform that powers synchronized digital experiences in realtime. Whether attending an event in a virtual venue, receiving realtime financial information, or monitoring live car performance data – consumers simply expect realtime digital experiences as standard. Ably provides a suite of APIs to build, extend, and deliver powerful digital experiences in realtime for more than 250 million devices across 80 countries each month. Organizations like Bloomberg, HubSpot, Verizon, and Hopin depend on Ably’s platform to offload the growing complexity of business-critical realtime data synchronization at global scale. For more information, see the [Ably documentation](https://ably.com/documentation)._
 
-This is a Ruby client library for Ably. The library currently targets the [Ably 1.2 client library specification](https://ably.com/documentation/client-lib-development-guide/features/). You can see the complete list of features this client library supports in [our client library SDKs feature support matrix](https://ably.com/download/sdk-feature-support-matrix).
+This is a Ruby client library for Ably. The library currently targets the [Ably 2.0.0 client library specification](https://ably.com/documentation/client-lib-development-guide/features/). You can see the complete list of features this client library supports in [our client library SDKs feature support matrix](https://ably.com/download/sdk-feature-support-matrix).
 
 ## Supported platforms
 

lib/ably/models/protocol_message.rb

@@ -20,8 +20,6 @@ module Ably::Models
   #   @return [String] Contains a serial number for a message on the current channel
   # @!attribute [r] connection_id
   #   @return [String] Contains a string private connection key used to recover this connection
-  # @!attribute [r] connection_serial
-  #   @return [Bignum] Contains a serial number for a message sent from the server to the client
   # @!attribute [r] message_serial
   #   @return [Bignum] Contains a serial number for a message sent from the client to the server
   # @!attribute [r] timestamp
@@ -129,12 +127,6 @@ def message_serial
       raise TypeError, "msg_serial '#{attributes[:msg_serial]}' is invalid, a positive Integer is expected for a ProtocolMessage"
     end
 
-    def connection_serial
-      Integer(attributes[:connection_serial])
-    rescue TypeError
-      raise TypeError, "connection_serial '#{attributes[:connection_serial]}' is invalid, a positive Integer is expected for a ProtocolMessage"
-    end
-
     def count
       [1, attributes[:count].to_i].max
     end
@@ -146,26 +138,12 @@ def has_message_serial?
       false
     end
 
-    # @api private
-    def has_connection_serial?
-      connection_serial && true
+    def has_channel_serial?
+      channel_serial && true
     rescue TypeError
       false
     end
 
-    def serial
-      if has_connection_serial?
-        connection_serial
-      else
-        message_serial
-      end
-    end
-
-    # @api private
-    def has_serial?
-      has_connection_serial? || has_message_serial?
-    end
-
     def messages
       @messages ||=
         Array(attributes[:messages]).map do |message|
@@ -271,7 +249,7 @@ def attributes
     # Return a JSON ready object from the underlying #attributes using Ably naming conventions for keys
     def as_json(*args)
       raise TypeError, ':action is missing, cannot generate a valid Hash for ProtocolMessage' unless action
-      raise TypeError, ':msg_serial or :connection_serial is missing, cannot generate a valid Hash for ProtocolMessage' if ack_required? && !has_serial?
+      raise TypeError, ':msg_serial is missing, cannot generate a valid Hash for ProtocolMessage' if ack_required? && !has_message_serial?
 
       attributes.dup.tap do |hash_object|
         hash_object['action']   = action.to_i
@@ -296,11 +274,12 @@ def to_s
     end
 
     # True if the ProtocolMessage appears to be invalid, however this is not a guarantee
+    # Used for validating incoming protocol messages, so no need to add unnecessary checks
     # @return [Boolean]
     # @api private
     def invalid?
       action_enum = action rescue nil
-      !action_enum || (ack_required? && !has_serial?)
+      !action_enum
     end
 
     # @!attribute [r] logger

lib/ably/modules/safe_deferrable.rb

@@ -39,7 +39,7 @@ def errback(&block)
       end
     end
 
-    # Mark the Deferrable as succeeded and trigger all callbacks.
+    # Mark the Deferrable as succeeded and trigger all success callbacks.
     # See http://www.rubydoc.info/gems/eventmachine/1.0.7/EventMachine/Deferrable#succeed-instance_method
     #
     # @return [void]
@@ -48,7 +48,7 @@ def succeed(*args)
       super(*args)
     end
 
-    # Mark the Deferrable as failed and trigger all callbacks.
+    # Mark the Deferrable as failed and trigger all error callbacks.
     # See http://www.rubydoc.info/gems/eventmachine/1.0.7/EventMachine/Deferrable#fail-instance_method
     #
     # @return [void]

lib/ably/modules/state_emitter.rb

@@ -3,7 +3,7 @@ module Ably::Modules
   # the instance variable @state is used exclusively, the {Enum} STATE is defined prior to inclusion of this
   # module, and the class is an {EventEmitter}.  It then emits state changes.
   #
-  # It also ensures the EventEmitter is configured to retrict permitted events to the
+  # It also ensures the EventEmitter is configured to restrict permitted events to the
   # the available STATEs or EVENTs if defined i.e. if EVENTs includes an additional type such as
   # :update, then it will support all EVENTs being emitted. EVENTs must be a superset of STATEs
   #

lib/ably/realtime/channel.rb

@@ -42,7 +42,7 @@ class Channel
       #
       # @spec RTL2b
       #
-      # The permited states for this channel
+      # The permitted states for this channel
       STATE = ruby_enum('STATE',
         :initialized,
         :attaching,

lib/ably/realtime/channel/channel_manager.rb

@@ -209,6 +209,7 @@ def send_attach_protocol_message
           message_options[:flags] = message_options[:flags].to_i | Ably::Models::ProtocolMessage::ATTACH_FLAGS_MAPPING[:resume]
         end
 
+        message_options[:channelSerial] = channel.properties.channel_serial # RTL4c1
         send_state_change_protocol_message Ably::Models::ProtocolMessage::ACTION.Attach, :suspended, message_options
       end
 

lib/ably/realtime/channel/channel_properties.rb

@@ -18,6 +18,15 @@ class ChannelProperties
       #
       attr_reader :attach_serial
 
+      # ChannelSerial contains the channelSerial from latest ProtocolMessage of action type
+      # Message/PresenceMessage received on the channel.
+      #
+      # @spec CP2b, RTL15b
+      #
+      # @return [String]
+      #
+      attr_accessor :channel_serial
+
       def initialize(channel)
         @channel = channel
       end

lib/ably/realtime/channel/channel_state_machine.rb

@@ -55,6 +55,7 @@ class ChannelStateMachine
       end
 
       after_transition(to: [:detached, :failed, :suspended]) do |channel, current_transition|
+        channel.properties.channel_serial = nil # RTP5a1
         err = error_from_state_change(current_transition)
         channel.manager.fail_queued_messages(err) if channel.failed? or channel.suspended? #RTL11
         channel.manager.log_channel_error err if err

lib/ably/realtime/channels.rb

@@ -46,6 +46,26 @@ def release(channel)
           @channels.delete(channel)
         end if @channels.has_key?(channel)
       end
+
+      # Sets channel serial to each channel from given serials hashmap
+      # @param [Hash] serials - map of channel name to respective channel serial
+      # @api private
+      def set_channel_serials(serials)
+        serials.each do |channel_name, channel_serial|
+          get(channel_name).properties.channel_serial = channel_serial
+        end
+      end
+
+      # @return [Hash] serials - map of channel name to respective channel serial
+      # @api private
+      def get_channel_serials
+        channel_serials = {}
+        self.each do |channel|
+          channel_serials[channel.name] = channel.properties.channel_serial if channel.state == :attached
+        end
+        channel_serials
+      end
+
     end
   end
 end

lib/ably/realtime/client.rb

@@ -1,5 +1,6 @@
 require 'uri'
 require 'ably/realtime/channel/publisher'
+require 'ably/realtime/recovery_key_context'
 
 module Ably
   module Realtime
@@ -11,6 +12,7 @@ class Client
       include Ably::Modules::Conversions
 
       extend Forwardable
+      using Ably::Util::AblyExtensions
 
       DOMAIN = 'realtime.ably.io'
 
@@ -120,17 +122,23 @@ def initialize(options)
           acc[key.to_s] = value.to_s
         end
         @rest_client           = Ably::Rest::Client.new(options.merge(realtime_client: self))
-        @echo_messages         = rest_client.options.fetch(:echo_messages, true) == false ? false : true
-        @queue_messages        = rest_client.options.fetch(:queue_messages, true) == false ? false : true
+        @echo_messages         = rest_client.options.fetch_with_default(:echo_messages, true)
+        @queue_messages        = rest_client.options.fetch_with_default(:queue_messages, true)
         @custom_realtime_host  = rest_client.options[:realtime_host] || rest_client.options[:ws_host]
-        @auto_connect          = rest_client.options.fetch(:auto_connect, true) == false ? false : true
-        @recover               = rest_client.options[:recover]
-
-        raise ArgumentError, "Recovery key '#{recover}' is invalid" if recover && !recover.match(Connection::RECOVER_REGEX)
+        @auto_connect          = rest_client.options.fetch_with_default(:auto_connect, true)
+        @recover               = rest_client.options.fetch_with_default(:recover, '')
 
         @auth       = Ably::Realtime::Auth.new(self)
         @channels   = Ably::Realtime::Channels.new(self)
         @connection = Ably::Realtime::Connection.new(self, options)
+
+        unless @recover.nil_or_empty?
+          recovery_context = RecoveryKeyContext.from_json(@recover, logger)
+          unless recovery_context.nil?
+            @channels.set_channel_serials recovery_context.channel_serials # RTN16j
+            @connection.set_msg_serial_from_recover = recovery_context.msg_serial  # RTN16f
+          end
+        end
       end
 
       # Return a {Ably::Realtime::Channel Realtime Channel} for the given name

lib/ably/realtime/client/incoming_message_dispatcher.rb

@@ -43,19 +43,24 @@ def dispatch_protocol_message(*args)
           raise ArgumentError, "Expected a ProtocolMessage. Received #{protocol_message}"
         end
 
-        unless protocol_message.action.match_any?(:nack, :error)
-          logger.debug { "#{protocol_message.action} received: #{protocol_message}" }
+        # RTL15b
+        if protocol_message.has_channel_serial? &&
+          (
+            protocol_message.action == :message ||
+              protocol_message.action == :presence ||
+              protocol_message.action == :attached
+          )
+          get_channel(protocol_message.channel).tap do |channel|
+            logger.info "Setting channel serial for channel #{channel.name}, " <<
+                          "Previous: #{channel.properties.channel_serial}, New: #{protocol_message.channel_serial}"
+            channel.properties.channel_serial = protocol_message.channel_serial
+          end
         end
 
-        if protocol_message.action.match_any?(:sync, :presence, :message)
-          if connection.serial && protocol_message.has_connection_serial? && protocol_message.connection_serial <= connection.serial
-            error_message = "Protocol error, duplicate message received for serial #{protocol_message.connection_serial}"
-            logger.error error_message
-            return
-          end
+        unless protocol_message.action.match_any?(:nack, :error)
+          logger.debug { "#{protocol_message.action} received: #{protocol_message}" }
         end
 
-        update_connection_recovery_info protocol_message
         connection.set_connection_confirmed_alive
 
         case protocol_message.action
@@ -172,10 +177,6 @@ def process_connected_update_message(protocol_message)
         end
       end
 
-      def update_connection_recovery_info(protocol_message)
-        connection.update_connection_serial protocol_message.connection_serial if protocol_message.has_connection_serial?
-      end
-
       def ack_pending_queue_for_message_serial(ack_protocol_message)
         drop_pending_queue_from_ack(ack_protocol_message) do |protocol_message|
           ack_messages protocol_message.messages