Feature/no connection serial recovery key

lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java

@@ -9,14 +9,10 @@
 import io.ably.lib.rest.AblyRest;
 import io.ably.lib.rest.Auth;
 import io.ably.lib.transport.ConnectionManager;
-import io.ably.lib.types.AblyException;
-import io.ably.lib.types.ChannelOptions;
-import io.ably.lib.types.ClientOptions;
-import io.ably.lib.types.ErrorInfo;
-import io.ably.lib.types.ProtocolMessage;
-import io.ably.lib.types.ReadOnlyMap;
+import io.ably.lib.types.*;
 import io.ably.lib.util.InternalMap;
 import io.ably.lib.util.Log;
+import io.ably.lib.util.StringUtils;
 
 /**
  * A client that extends the functionality of the {@link AblyRest} and provides additional realtime-specific features.
@@ -71,6 +67,14 @@ public void onConnectionStateChanged(ConnectionStateListener.ConnectionStateChan
             }
         });
 
+        if (!StringUtils.isNullOrEmpty(options.recover)) {
+            RecoveryKeyContext recoveryKeyContext = RecoveryKeyContext.decode(options.recover);
+            if (recoveryKeyContext != null) {
+                setChannelSerialsFromRecoverOption(recoveryKeyContext.getChannelSerials());
+                connection.connectionManager.msgSerial = recoveryKeyContext.getMsgSerial(); //RTN16f
+            }
+        }
+
         if(options.autoConnect) connection.connect();
     }
 
@@ -274,6 +278,25 @@ private void clear() {
         }
     }
 
+    protected void setChannelSerialsFromRecoverOption(Map<String, String> serials) {
+        for (Map.Entry<String, String> entry : serials.entrySet()) {
+            String channelName = entry.getKey();
+            String channelSerial = entry.getValue();
+            Channel channel = this.channels.get(channelName);
+            if (channel != null) {
+                channel.properties.channelSerial = channelSerial;
+            }
+        }
+    }
+
+    protected Map<String, String> getChannelSerials() {
+        Map<String, String> channelSerials = new HashMap<>();
+        for (Channel channel : this.channels.values()) {
+            channelSerials.put(channel.name, channel.properties.channelSerial);
+        }
+        return channelSerials;
+    }
+
     /********************
      * internal
      ********************/

lib/src/main/java/io/ably/lib/realtime/ChannelBase.java

@@ -37,6 +37,7 @@
 import io.ably.lib.util.EventEmitter;
 import io.ably.lib.util.Log;
 import io.ably.lib.util.ReconnectionStrategy;
+import io.ably.lib.util.StringUtils;
 
 /**
  * Enables messages to be published and subscribed to.
@@ -131,6 +132,11 @@ private void setState(ChannelState newState, ErrorInfo reason, boolean resumed,
             this.retryCount = 0;
         }
 
+        // RTP5a1
+        if (newState == ChannelState.detached || newState == ChannelState.suspended || newState == ChannelState.failed) {
+            properties.channelSerial = null;
+        }
+
         if(notifyStateChange) {
             /* broadcast state change */
             emit(newState, stateChange);
@@ -248,8 +254,9 @@ private void attachImpl(final boolean forceReattach, final CompletionListener li
             }
         }
         if(this.decodeFailureRecoveryInProgress) {
-            attachMessage.channelSerial = this.lastPayloadProtocolMessageChannelSerial;
+            Log.v(TAG, "attach(); message decode recovery in progress.");
         }
+        attachMessage.channelSerial = properties.channelSerial;
         try {
             if (listener != null) {
                 on(new ChannelStateCompletionListener(listener, ChannelState.attached, ChannelState.failed));
@@ -325,7 +332,10 @@ private void detachImpl(CompletionListener listener) throws AblyException {
         if(!connectionManager.isActive())
             throw AblyException.fromErrorInfo(connectionManager.getStateErrorInfo());
 
-        /* send detach request */
+        sendDetachMessage(listener);
+    }
+
+    private void sendDetachMessage(CompletionListener listener) throws AblyException {
         ProtocolMessage detachMessage = new ProtocolMessage(Action.detach, this.name);
         try {
             if (listener != null) {
@@ -338,7 +348,7 @@ private void detachImpl(CompletionListener listener) throws AblyException {
             } else {
                 setState(ChannelState.detaching, null);
             }
-            connectionManager.send(detachMessage, true, null);
+            ably.connection.connectionManager.send(detachMessage, true, null);
         } catch(AblyException e) {
             throw e;
         }
@@ -399,7 +409,15 @@ private void setAttached(ProtocolMessage message) {
             Log.v(TAG, String.format(Locale.ROOT, "Server initiated attach for channel %s", name));
             /* emit UPDATE event according to RTL12 */
             emitUpdate(null, resumed);
-        } else {
+        } else if (state == ChannelState.detaching || state == ChannelState.detached) { //RTL5k
+            Log.v(TAG, "setAttached(): channel is in detaching state, as per RTL5k sending detach message!");
+            try {
+                sendDetachMessage(null);
+            } catch (AblyException e) {
+                Log.e(TAG, e.getMessage(), e);
+            }
+        }
+        else {
             this.attachResume = true;
             setState(ChannelState.attached, message.error, resumed);
             presence.setAttached(message.hasFlag(Flag.has_presence), this.ably.connection.id);
@@ -850,7 +868,6 @@ private void onMessage(final ProtocolMessage protocolMessage) {
         }
 
         lastPayloadMessageId = lastMessage.id;
-        lastPayloadProtocolMessageChannelSerial = protocolMessage.channelSerial;
 
         for (final Message msg : messages) {
             this.listeners.onMessage(msg);
@@ -1264,6 +1281,15 @@ else if(stateChange.current.equals(failureState)) {
     }
 
     void onChannelMessage(ProtocolMessage msg) {
+        // RTL15b
+        if (!StringUtils.isNullOrEmpty(msg.channelSerial) && (msg.action == Action.message ||
+            msg.action == Action.presence || msg.action == Action.attached)) {
+            Log.v(TAG, String.format(
+                Locale.ROOT, "Setting channel serial for channelName - %s, previous - %s, current - %s",
+                name, properties.channelSerial, msg.channelSerial));
+            properties.channelSerial = msg.channelSerial;
+        }
+
         switch(msg.action) {
         case attached:
             setAttached(msg);
@@ -1369,7 +1395,6 @@ public void once(ChannelState state, ChannelStateListener listener) {
      */
     private Set<ChannelMode> modes;
     private String lastPayloadMessageId;
-    private String lastPayloadProtocolMessageChannelSerial;
     private boolean decodeFailureRecoveryInProgress;
     private final DecodingContext decodingContext;
 }

lib/src/main/java/io/ably/lib/realtime/Connection.java

@@ -4,6 +4,7 @@
 import io.ably.lib.transport.ConnectionManager;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ErrorInfo;
+import io.ably.lib.types.RecoveryKeyContext;
 import io.ably.lib.util.EventEmitter;
 import io.ably.lib.util.Log;
 import io.ably.lib.util.PlatformAgentProvider;
@@ -49,25 +50,36 @@ public class Connection extends EventEmitter<ConnectionEvent, ConnectionStateLis
      * for more information.
      * <p>
      * Spec: RTN16b, RTN16c
+     * @deprecated use createRecoveryKey method instead.
      */
+    @Deprecated
     public String recoveryKey;
 
     /**
-     * A unique public identifier for this connection, used to identify this member.
-     * <p>
-     * Spec: RTN8
+     * Spec: RTN16g
+     *
+     * @return a json string which incorporates the @connectionKey@, the current @msgSerial@,
+     * and a collection of pairs of channel @name@ and current @channelSerial@ for every currently attached channel.
      */
-    public String id;
+    public String createRecoveryKey() {
+        if (key == null || key.isEmpty() || this.state == ConnectionState.closing ||
+            this.state == ConnectionState.closed ||
+            this.state == ConnectionState.failed ||
+            this.state == ConnectionState.suspended
+        ) {
+            //RTN16h
+            return null;
+        }
+
+        return new RecoveryKeyContext(key, connectionManager.msgSerial, ably.getChannelSerials()).encode();
+    }
 
     /**
-     * The serial number of the last message to be received on this connection,
-     * used automatically by the library when recovering or resuming a connection.
-     * When recovering a connection explicitly, the recoveryKey is used in the recover
-     * client options as it contains both the key and the last message serial.
+     * A unique public identifier for this connection, used to identify this member.
      * <p>
-     * Spec: RTN10
+     * Spec: RTN8
      */
-    public long serial;
+    public String id;
 
     /**
      * Explicitly calling connect() is unnecessary unless the autoConnect attribute of the {@link io.ably.lib.types.ClientOptions}

lib/src/main/java/io/ably/lib/transport/ConnectionManager.java

@@ -34,6 +34,7 @@
 import io.ably.lib.util.Log;
 import io.ably.lib.util.PlatformAgentProvider;
 import io.ably.lib.util.ReconnectionStrategy;
+import io.ably.lib.util.StringUtils;
 
 public class ConnectionManager implements ConnectListener {
     final ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
@@ -1193,19 +1194,19 @@ public void onMessage(ITransport transport, ProtocolMessage message) throws Ably
     }
 
     private void onChannelMessage(ProtocolMessage message) {
-        if(message.connectionSerial != null) {
-            connection.serial = message.connectionSerial.longValue();
-            if (connection.key != null)
-                connection.recoveryKey = connection.key + ":" + message.connectionSerial;
-        }
         channels.onMessage(message);
+        connection.recoveryKey = connection.createRecoveryKey();
     }
 
     private synchronized void onConnected(ProtocolMessage message) {
         final ErrorInfo error = message.error;
         boolean reattachOnResumeFailure = false; // this will indicate that channel must reattach when connected
         // event is received
 
+        boolean isConnectionResumeOrRecoverAttempt = !StringUtils.isNullOrEmpty(connection.key) ||
+            !StringUtils.isNullOrEmpty(ably.options.recover);
+        ably.options.recover = null; // RTN16k, explicitly setting null, so it won't be used for subsequent connection requests
+
         connection.reason = error;
         if (connection.id != null) { // there was a previous connection, so this is a resume and RTN15c applies
             Log.d(TAG, "There was a connection resume");
@@ -1240,12 +1241,6 @@ private synchronized void onConnected(ProtocolMessage message) {
 
         connection.id = message.connectionId;
 
-        if(message.connectionSerial != null) {
-            connection.serial = message.connectionSerial;
-            if (connection.key != null)
-                connection.recoveryKey = connection.key + ":" + message.connectionSerial;
-        }
-
         ConnectionDetails connectionDetails = message.connectionDetails;
         /* Get any parameters from connectionDetails. */
         connection.key = connectionDetails.connectionKey; //RTN16d
@@ -1260,6 +1255,9 @@ private synchronized void onConnected(ProtocolMessage message) {
             requestState(transport, new StateIndication(ConnectionState.failed, e.errorInfo));
             return;
         }
+
+        connection.recoveryKey = connection.createRecoveryKey();
+
         /* indicated connected currentState */
         final StateIndication stateIndication = new StateIndication(ConnectionState.connected, error, null, null,
             reattachOnResumeFailure);
@@ -1504,7 +1502,6 @@ private class ConnectParams extends TransportParams {
         ConnectParams(ClientOptions options, PlatformAgentProvider platformAgentProvider) {
             super(options, platformAgentProvider);
             this.connectionKey = connection.key;
-            this.connectionSerial = String.valueOf(connection.serial);
             this.port = Defaults.getPort(options);
         }
     }
@@ -1905,7 +1902,7 @@ private boolean isFatalError(ErrorInfo err) {
     private boolean suppressRetry; /* for tests only; modified via reflection */
     private ITransport transport;
     private long suspendTime;
-    private long msgSerial;
+    public long msgSerial;
     private long lastActivity;
     private CMConnectivityListener connectivityListener;
     private long connectionStateTtl = Defaults.connectionStateTtl;

lib/src/main/java/io/ably/lib/transport/ITransport.java

@@ -1,21 +1,16 @@
 package io.ably.lib.transport;
 
-import io.ably.lib.types.AblyException;
-import io.ably.lib.types.ClientOptions;
-import io.ably.lib.types.ErrorInfo;
-import io.ably.lib.types.Param;
-import io.ably.lib.types.ProtocolMessage;
+import io.ably.lib.types.*;
 import io.ably.lib.util.AgentHeaderCreator;
 import io.ably.lib.util.Log;
 import io.ably.lib.util.PlatformAgentProvider;
+import io.ably.lib.util.StringUtils;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 
 import java.util.List;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 public interface ITransport {
 
@@ -39,7 +34,6 @@ class TransportParams {
         protected String host;
         protected int port;
         protected String connectionKey;
-        protected String connectionSerial;
         protected Mode mode;
         protected boolean heartbeats;
         private final PlatformAgentProvider platformAgentProvider;
@@ -68,20 +62,14 @@ public Param[] getConnectParams(Param[] baseParams) {
             paramList.add(new Param("format", (options.useBinaryProtocol ? "msgpack" : "json")));
             if(!options.echoMessages)
                 paramList.add(new Param("echo", "false"));
-            if(connectionKey != null) {
+            if(!StringUtils.isNullOrEmpty(connectionKey)) {
                 mode = Mode.resume;
                 paramList.add(new Param("resume", connectionKey));
-                if(connectionSerial != null)
-                    paramList.add(new Param("connectionSerial", connectionSerial));
-            } else if(options.recover != null) {
+            } else if(!StringUtils.isNullOrEmpty(options.recover)) { // RTN16k
                 mode = Mode.recover;
-                Pattern recoverSpec = Pattern.compile("^([\\w\\-\\!]+):(\\-?\\d+)$");
-                Matcher match = recoverSpec.matcher(options.recover);
-                if(match.matches()) {
-                    paramList.add(new Param("recover", match.group(1)));
-                    paramList.add(new Param("connectionSerial", match.group(2)));
-                } else {
-                    Log.e(TAG, "Invalid recover string specified");
+                RecoveryKeyContext recoveryKeyContext = RecoveryKeyContext.decode(options.recover);
+                if (recoveryKeyContext != null) {
+                    paramList.add(new Param("recover", recoveryKeyContext.getConnectionKey()));
                 }
             }
             if(options.clientId != null)

lib/src/main/java/io/ably/lib/types/ChannelProperties.java

@@ -13,5 +13,13 @@ public class ChannelProperties {
      */
     public String attachSerial;
 
+    /**
+     * ChannelSerial contains the channelSerial from latest ProtocolMessage of action type
+     * Message/PresenceMessage received on the channel.
+     * <p>
+     * Spec: CP2b, RTL15b
+     */
+    public String channelSerial;
+
     public ChannelProperties() {}
 }

lib/src/main/java/io/ably/lib/types/ProtocolMessage.java

@@ -98,7 +98,6 @@ public ProtocolMessage(Action action, String channel) {
     public String channel;
     public String channelSerial;
     public String connectionId;
-    public Long connectionSerial;
     public Long msgSerial;
     public long timestamp;
     public Message[] messages;
@@ -198,9 +197,6 @@ ProtocolMessage readMsgpack(MessageUnpacker unpacker) throws IOException {
                 case "connectionId":
                     connectionId = unpacker.unpackString();
                     break;
-                case "connectionSerial":
-                    connectionSerial = Long.valueOf(unpacker.unpackLong());
-                    break;
                 case "msgSerial":
                     msgSerial = Long.valueOf(unpacker.unpackLong());
                     break;