[ECO-4687]Feature/resume recover

ably.gemspec

@@ -5,7 +5,7 @@ require 'ably/version'
 
 Gem::Specification.new do |spec|
   spec.name          = 'ably'
-  spec.version       = Ably::VERSION
+  spec.version       = Ably::LIB_VERSION
   spec.authors       = ['Lewis Marshall', "Matthew O'Riordan"]
   spec.email         = ['[email protected]', '[email protected]']
   spec.description   = %q{A Ruby client library for ably.io realtime messaging}

lib/ably/agent.rb

@@ -1,3 +1,3 @@
 module Ably
-  AGENT = "ably-ruby/#{Ably::VERSION} ruby/#{RUBY_VERSION}"
+  AGENT = "ably-ruby/#{Ably::LIB_VERSION} ruby/#{RUBY_VERSION}"
 end

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,41 +127,18 @@ 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
 
     # @api private
     def has_message_serial?
-      message_serial && true
-    rescue TypeError
-      false
-    end
-
-    # @api private
-    def has_connection_serial?
-      connection_serial && true
-    rescue TypeError
-      false
-    end
-
-    def serial
-      if has_connection_serial?
-        connection_serial
-      else
-        message_serial
-      end
+      not Ably::Util::String::is_null_or_empty(message_serial)
     end
 
     # @api private
     def has_serial?
-      has_connection_serial? || has_message_serial?
+      has_message_serial?
     end
 
     def messages
@@ -271,7 +246,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_serial?
 
       attributes.dup.tap do |hash_object|
         hash_object['action']   = action.to_i

lib/ably/modules/http_helpers.rb

@@ -18,7 +18,7 @@ def encode64(text)
     end
 
     def user_agent
-      "Ably Ruby client #{Ably::VERSION} (https://www.ably.io)"
+      "Ably Ruby client #{Ably::LIB_VERSION} (https://www.ably.io)"
     end
 
     def setup_outgoing_middleware(builder)

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,22 @@ def release(channel)
           @channels.delete(channel)
         end if @channels.has_key?(channel)
       end
+
+      # @param [Hash] serials - map of channel name to respective serial
+      def set_channel_serials(serials)
+        serials.each do |channel_name, channel_serial|
+          channels[channel_name].properties.channel_serial = channel_serial
+        end
+      end
+
+      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

@@ -62,7 +62,7 @@ class Client
 
       # When a recover option is specified a connection inherits the state of a previous connection that may have existed under a different instance of the Realtime library, please refer to the API documentation for further information on connection state recovery
       # @return [String,Nil]
-      attr_reader :recover
+      attr_accessor :recover
 
       # Additional parameters to be sent in the querystring when initiating a realtime connection
       # @return [Hash]
@@ -120,17 +120,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(:echo_messages, true)
+        @queue_messages        = rest_client.options.fetch(: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(:auto_connect, true)
+        @recover               = rest_client.options.fetch(:recover, '')
 
         @auth       = Ably::Realtime::Auth.new(self)
         @channels   = Ably::Realtime::Channels.new(self)
         @connection = Ably::Realtime::Connection.new(self, options)
+
+        unless @recover.empty?
+          recovery_context = RecoveryKeyContext.from_json(@recover, logger)
+          unless recovery_context.nil?
+            @channels.set_channel_serials recovery_context.channel_serials # RTN16j
+            @connection.message_serial = recovery_context.msg_serial  # RTN16f
+          end
+        end
       end
 
       # Return a {Ably::Realtime::Channel Realtime Channel} for the given name
@@ -302,13 +308,6 @@ def logger
         @logger ||= Ably::Logger.new(self, log_level, rest_client.logger.custom_logger)
       end
 
-      # Disable connection recovery, typically used after a connection has been recovered
-      # @return [void]
-      # @api private
-      def disable_automatic_connection_recovery
-        @recover = nil
-      end
-
       # @!attribute [r] fallback_endpoint
       # @return [URI::Generic] Fallback endpoint used to connect to the realtime Ably service. Note, after each connection attempt, a new random {Ably::FALLBACK_HOSTS fallback host} or provided fallback hosts are used
       # @api private

lib/ably/realtime/client/incoming_message_dispatcher.rb

@@ -38,6 +38,22 @@ def logger
 
       def dispatch_protocol_message(*args)
         protocol_message = args.first
+        # RTL15b
+        unless protocol_message.nil?
+          if protocol_message.has_message_serial? &&
+                  (
+                    protocol_message.action == :message ||
+                    protocol_message.action == :presence ||
+                    protocol_message.action == :attached
+                  )
+
+            logger.info "Setting channel serial for #{channel.name}"
+            logger.info "Previous serial #{channel.name}, new serial #{protocol_message.channel_serial}"
+            get_channel(protocol_message.channel).tap do |channel|
+              channel.properties.channel_serial = protocol_message.channel_serial
+            end
+          end
+        end
 
         unless protocol_message.kind_of?(Ably::Models::ProtocolMessage)
           raise ArgumentError, "Expected a ProtocolMessage. Received #{protocol_message}"
@@ -47,15 +63,6 @@ def dispatch_protocol_message(*args)
           logger.debug { "#{protocol_message.action} received: #{protocol_message}" }
         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
-        end
-
-        update_connection_recovery_info protocol_message
         connection.set_connection_confirmed_alive
 
         case protocol_message.action
@@ -172,10 +179,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