Protocol 2.0

android/src/androidTest/java/io/ably/lib/test/android/AndroidPushTest.java

@@ -53,7 +53,6 @@
 import io.ably.lib.types.ClientOptions;
 import io.ably.lib.types.ErrorInfo;
 import io.ably.lib.types.Param;
-import io.ably.lib.types.RegistrationToken;
 import io.ably.lib.util.Base64Coder;
 import io.ably.lib.util.IntentUtils;
 import io.ably.lib.util.JsonUtils;
@@ -1367,11 +1366,12 @@ public void run() throws Exception {
                 }
 
                 testActivation.registerAndWait();
-                DeviceDetails otherDevice = DeviceDetails.fromJsonObject(JsonUtils.object()
+                LocalDevice otherDevice = LocalDevice.fromJsonObject(JsonUtils.object()
                         .add("id", "other")
                         .add("platform", "android")
                         .add("formFactor", "tablet")
                         .add("metadata", JsonUtils.object())
+                        .add("deviceSecret", "testdevicesecret==") //Required by API protocol 2.0
                         .add("push", JsonUtils.object()
                                 .add("recipient", JsonUtils.object()
                                         .add("transportType", "fcm")

android/src/main/java/io/ably/lib/push/ActivationStateMachine.java

@@ -19,7 +19,6 @@
 import io.ably.lib.types.Callback;
 import io.ably.lib.types.ErrorInfo;
 import io.ably.lib.types.Param;
-import io.ably.lib.types.RegistrationToken;
 import io.ably.lib.util.IntentUtils;
 import io.ably.lib.util.Log;
 import io.ably.lib.util.ParamsUtils;

android/src/main/java/io/ably/lib/push/LocalDevice.java

@@ -15,9 +15,10 @@
 import io.ably.lib.types.RegistrationToken;
 import io.ably.lib.util.Base64Coder;
 import io.ably.lib.util.Log;
+import io.ably.lib.util.Serialisation;
 
 public class LocalDevice extends DeviceDetails {
-    public String deviceSecret;
+
     public String deviceIdentityToken;
     private final Storage storage;
 
@@ -35,12 +36,11 @@ public LocalDevice(ActivationContext activationContext, Storage storage) {
     }
 
     public JsonObject toJsonObject() {
-        JsonObject o = super.toJsonObject();
-        if (deviceSecret != null) {
-            o.addProperty("deviceSecret", deviceSecret);
-        }
+        return super.toJsonObject();
+    }
 
-        return o;
+    public static LocalDevice fromJsonObject(JsonObject o) {
+        return Serialisation.gson.fromJson(o, LocalDevice.class);
     }
 
     private void loadPersisted() {

core/src/main/java/io/ably/lib/realtime/AblyRealtimeBase.java

@@ -58,7 +58,7 @@ public AblyRealtimeBase(String key, PlatformAgentProvider platformAgentProvider)
      * @param platformAgentProvider for providing the platform specific part of the agent header
      * @throws AblyException
      */
-    public AblyRealtimeBase(ClientOptions options, PlatformAgentProvider platformAgentProvider) throws AblyException {
+    public AblyRealtimeBase(final ClientOptions options, PlatformAgentProvider platformAgentProvider) throws AblyException {
         super(options, platformAgentProvider);
         final InternalChannels channels = new InternalChannels();
         this.channels = (Channels<ChannelType>) channels;
@@ -72,6 +72,31 @@ public void onConnectionStateChanged(ConnectionStateListener.ConnectionStateChan
             }
         });
 
+        if (options.recover != null) {
+            ConnectionRecoveryKey recoveryKey = ConnectionRecoveryKey.fromJson(options.recover);
+            if (recoveryKey == null) {
+                Log.d(TAG, "Recovery key initialization failed!");
+                connection.connectionManager.msgSerial = 0; //RTN16f
+            } else {
+                connection.connectionManager.msgSerial = recoveryKey.msgSerial; //RTN16f
+
+                for (Map.Entry<String, String> entry : recoveryKey.serials.entrySet()) {
+                    //RTN16j
+                    RealtimeChannelBase channel = channels.get(entry.getKey());
+                    if (channel != null) {
+                        channel.properties.channelSerial = entry.getValue(); //RTN16i
+                    }
+                }
+            }
+
+            connection.on(ConnectionEvent.connected, new ConnectionStateListener() {
+                @Override
+                public void onConnectionStateChanged(ConnectionStateChange state) {
+                    options.recover = null; //RTN16k
+                }
+            });
+        }
+
         if(options.autoConnect) connection.connect();
     }
 
@@ -183,6 +208,21 @@ public void suspendAll(ErrorInfo error, boolean notifyStateChange) {
             }
         }
 
+        /**
+         * By spec RTN15c6, RTN15c7
+         */
+        @Override
+        public void reAttach() {
+            for (Map.Entry<String, RealtimeChannelBase> entry : map.entrySet()) {
+                RealtimeChannelBase channel = entry.getValue();
+                if (channel.state == ChannelState.attaching || channel.state == ChannelState.attached || channel.state == ChannelState.suspended) {
+                    Log.d(TAG, "reAttach(); channel = " + channel.name);
+                    channel.state = ChannelState.attaching;
+                    channel.attach(true, null);
+                }
+            }
+        }
+
         private void clear() {
             map.clear();
         }

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

@@ -29,23 +29,12 @@ public class Connection extends EventEmitter<ConnectionEvent, ConnectionStateLis
      */
     public String key;
 
-    /**
-     * RTN16b) Connection#recoveryKey is an attribute composed of the connection key and latest
-     * serial received on the connection
-     */
-    public String recoveryKey;
-
     /**
      * A public identifier for this connection, used to identify
      * this member in presence events and message ids.
      */
     public String id;
 
-    /**
-     * The serial number of the last message to be received on this connection.
-     */
-    public long serial;
-
     /**
      * Causes the library to re-attempt connection, if it was previously explicitly
      * closed by the user, or was closed as a result of an unrecoverable error.
@@ -56,6 +45,7 @@ public void connect() {
 
     /**
      * Send a heartbeat message to the Ably service and await a response.
+     *
      * @param listener a listener to be notified of the outcome of this message.
      */
     public void ping(CompletionListener listener) {
@@ -69,7 +59,6 @@ public void ping(CompletionListener listener) {
      */
     public void close() {
         key = null;
-        recoveryKey = null;
         connectionManager.close();
     }
 
@@ -92,7 +81,7 @@ public void onConnectionStateChange(ConnectionStateChange stateChange) {
     @Override
     protected void apply(ConnectionStateListener listener, ConnectionEvent event, Object... args) {
         try {
-            listener.onConnectionStateChanged((ConnectionStateChange)args[0]);
+            listener.onConnectionStateChanged((ConnectionStateChange) args[0]);
         } catch (Throwable t) {
             Log.e(TAG, "Unexpected exception calling ConnectionStateListener", t);
         }
@@ -103,6 +92,37 @@ public void emitUpdate(ErrorInfo errorInfo) {
             emit(ConnectionEvent.update, ConnectionStateListener.ConnectionStateChange.createUpdateEvent(errorInfo));
     }
 
+    /**
+     * 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 getRecoveryKey() {
+        if (key == null || connectionManager == null || connectionManager.getConnectionState() == null ||
+            connectionManager.getConnectionState().state == ConnectionState.closed ||
+            connectionManager.getConnectionState().state == ConnectionState.closing ||
+            connectionManager.getConnectionState().state == ConnectionState.failed ||
+            connectionManager.getConnectionState().state == ConnectionState.suspended
+        ) {
+            //RTN16h
+            return null;
+        }
+
+        ConnectionRecoveryKey recoveryKey = new ConnectionRecoveryKey();
+        recoveryKey.connectionKey = key;
+        recoveryKey.msgSerial = connectionManager.msgSerial;
+
+        for (Object channel : ably.channels.values()) {
+            if (channel instanceof RealtimeChannelBase) {
+                RealtimeChannelBase rcb = (RealtimeChannelBase) channel;
+                recoveryKey.serials.put(rcb.name, rcb.properties.channelSerial);
+            }
+        }
+
+        return recoveryKey.asJson();
+    }
+
     @Deprecated
     public void emit(ConnectionState state, ConnectionStateChange stateChange) {
         super.emit(state.getConnectionEvent(), stateChange);

core/src/main/java/io/ably/lib/realtime/ConnectionRecoveryKey.java

@@ -0,0 +1,34 @@
+package io.ably.lib.realtime;
+
+import com.google.gson.JsonSyntaxException;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import io.ably.lib.util.Serialisation;
+
+public class ConnectionRecoveryKey {
+
+    public String connectionKey;
+    public long msgSerial;
+    /**
+     * Key - channel name
+     * <p>
+     * Value - channelSerial
+     */
+    public Map<String, String> serials = new HashMap<>();
+
+    public String asJson() {
+        return Serialisation.gson.toJson(this);
+    }
+
+    public static ConnectionRecoveryKey fromJson(String json) {
+        try {
+            return Serialisation.gson.fromJson(json, ConnectionRecoveryKey.class);
+        } catch (JsonSyntaxException e) {
+            e.printStackTrace();
+            return null;
+        }
+    }
+
+}

core/src/main/java/io/ably/lib/realtime/Presence.java

@@ -295,6 +295,15 @@ public void onError(ErrorInfo reason) {
         }
     }
 
+    /**
+     * Spec: RTP17f, RTP17g
+     */
+    public void reEnter(ChannelState newState) {
+        if (newState == ChannelState.attached && channel.state == ChannelState.attached) {
+            internalPresence.reEnter();
+        }
+    }
+
     void setPresence(PresenceMessage[] messages, boolean broadcast, String syncChannelSerial) {
         Log.v(TAG, "setPresence(); channel = " + channel.name + "; broadcast = " + broadcast + "; syncChannelSerial = " + syncChannelSerial);
         String syncCursor = null;
@@ -872,7 +881,7 @@ synchronized Collection<PresenceMessage> get(Param[] params) throws AblyExceptio
          * false if the message is already superseded
          */
         synchronized boolean put(PresenceMessage item) {
-            String key = item.memberKey();
+            String key = item.clientId; //RTP17h
             /* we've seen this member, so do not remove it at the end of sync */
             if(residualMembers != null)
                 residualMembers.remove(key);
@@ -967,7 +976,7 @@ synchronized Collection<PresenceMessage> values(boolean wait) throws AblyExcepti
          * @return
          */
         synchronized boolean remove(PresenceMessage item) {
-            String key = item.memberKey();
+            String key = item.clientId; //RTP17h
             if (hasNewerItem(key, item))
                 return false;
             PresenceMessage existingItem = members.remove(key);
@@ -1031,6 +1040,21 @@ synchronized void clear() {
                 residualMembers.clear();
         }
 
+        /**
+         * Spec: RTP17g
+         */
+        synchronized void reEnter() {
+            for (Map.Entry<String, PresenceMessage> entry: members.entrySet()) {
+                PresenceMessage member = entry.getValue();
+                member.action = PresenceMessage.Action.enter;
+                try {
+                    updatePresence(member, null);
+                } catch (AblyException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+
         private boolean syncInProgress;
         private Collection<String> residualMembers;
         private final HashMap<String, PresenceMessage> members = new HashMap<String, PresenceMessage>();

core/src/main/java/io/ably/lib/realtime/RealtimeChannelBase.java

@@ -92,13 +92,21 @@ private void setState(ChannelState newState, ErrorInfo reason, boolean resumed)
     private void setState(ChannelState newState, ErrorInfo reason, boolean resumed, boolean notifyStateChange) {
         Log.v(TAG, "setState(): channel = " + name + "; setting " + newState);
         ChannelStateListener.ChannelStateChange stateChange;
-        synchronized(this) {
+        synchronized (this) {
             stateChange = new ChannelStateListener.ChannelStateChange(newState, this.state, reason, resumed);
             this.state = stateChange.current;
             this.reason = stateChange.reason;
         }
 
-        if(notifyStateChange) {
+        if (newState == ChannelState.detached || newState == ChannelState.suspended || newState == ChannelState.failed)
+            properties.channelSerial = null; //RTP5a1
+
+        if (newState == ChannelState.attached && state == ChannelState.attached) {
+            //RTP17f
+            presence.reEnter(newState);
+        }
+
+        if (notifyStateChange) {
             /* broadcast state change */
             emit(newState, stateChange);
         }
@@ -137,7 +145,7 @@ public void attach(CompletionListener listener) throws  AblyException {
         this.attach(false, listener);
     }
 
-    private void attach(boolean forceReattach, CompletionListener listener) {
+    void attach(boolean forceReattach, CompletionListener listener) {
         clearAttachTimers();
         attachWithTimeout(forceReattach, listener);
     }
@@ -168,6 +176,7 @@ private void attachImpl(final boolean forceReattach, final CompletionListener li
         /* send attach request and pending state */
         Log.v(TAG, "attach(); channel = " + name + "; sending ATTACH request");
         ProtocolMessage attachMessage = new ProtocolMessage(Action.attach, this.name);
+        attachMessage.channelSerial = this.properties.channelSerial;
         if(this.options != null) {
             if(this.options.hasParams()) {
                 attachMessage.params = CollectionUtils.copy(this.options.params);
@@ -177,7 +186,7 @@ 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.");
         }
         try {
             if (listener != null) {
@@ -301,6 +310,10 @@ private void setAttached(ProtocolMessage message) {
         properties.attachSerial = message.channelSerial;
         params = message.params;
         modes = ChannelMode.toSet(message.flags);
+        if (message.channelSerial != null) {
+            properties.channelSerial = message.channelSerial; //RTL4c1
+        }
+
         if(state == ChannelState.attached) {
             Log.v(TAG, String.format(Locale.ROOT, "Server initiated attach for channel %s", name));
             /* emit UPDATE event according to RTL12 */
@@ -727,7 +740,10 @@ private void onMessage(final ProtocolMessage protocolMessage) {
         }
 
         lastPayloadMessageId = lastMessage.id;
-        lastPayloadProtocolMessageChannelSerial = protocolMessage.channelSerial;
+
+        if (protocolMessage.channelSerial != null) {
+            properties.channelSerial = protocolMessage.channelSerial; //RTL15b
+        }
 
         for (final Message msg : messages) {
             this.listeners.onMessage(msg);
@@ -768,6 +784,9 @@ private void onPresence(ProtocolMessage message, String syncChannelSerial) {
             if(msg.timestamp == 0) msg.timestamp = message.timestamp;
             if(msg.id == null) msg.id = message.id + ':' + i;
         }
+        if (message.channelSerial != null) {
+            properties.channelSerial = message.channelSerial; //RTL15b
+        }
         presence.setPresence(messages, true, syncChannelSerial);
     }
 
@@ -902,6 +921,7 @@ public synchronized void publish(Message[] messages, CompletionListener listener
         switch(state) {
         case failed:
         case suspended:
+            properties.channelSerial = null; //RTP5a1
             throw AblyException.fromErrorInfo(new ErrorInfo("Unable to publish in failed or suspended state", 400, 40000));
         default:
             connectionManager.send(msg, queueMessages, listener);
@@ -1199,7 +1219,6 @@ public void once(ChannelState state, ChannelStateListener listener) {
     private Map<String, String> params;
     private Set<ChannelMode> modes;
     private String lastPayloadMessageId;
-    private String lastPayloadProtocolMessageChannelSerial;
     private boolean decodeFailureRecoveryInProgress;
     private final DecodingContext decodingContext;
 }