[ECO-5375] Refactored LiveObjects plugin to handle channelSerial

Author: sacOO7

lib/src/main/java/io/ably/lib/objects/LiveObjectsAdapter.java

@@ -0,0 +1,36 @@
+package io.ably.lib.objects;
+
+import io.ably.lib.plugins.PluginConnectionAdapter;
+import io.ably.lib.realtime.AblyRealtime;
+import io.ably.lib.realtime.CompletionListener;
+import io.ably.lib.types.AblyException;
+import io.ably.lib.types.ProtocolMessage;
+import io.ably.lib.util.Log;
+import org.jetbrains.annotations.NotNull;
+
+public interface LiveObjectsAdapter extends PluginConnectionAdapter {
+    void setChannelSerial(@NotNull String channelName, @NotNull String channelSerial);
+
+    class Adapter implements LiveObjectsAdapter {
+        private final AblyRealtime ably;
+        private static final String TAG = LiveObjectsAdapter.class.getName();
+
+        public Adapter(@NotNull AblyRealtime ably) {
+            this.ably = ably;
+        }
+
+        @Override
+        public void setChannelSerial(@NotNull String channelName, @NotNull String channelSerial) {
+            if (ably.channels.containsKey(channelName)) {
+                ably.channels.get(channelName).properties.channelSerial = channelSerial;
+            } else {
+                Log.e(TAG, "setChannelSerial(): channel not found: " + channelName);
+            }
+        }
+
+        @Override
+        public void send(ProtocolMessage msg, CompletionListener listener) throws AblyException {
+            ably.connection.connectionManager.send(msg, true, listener);
+        }
+    }
+}

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

@@ -6,8 +6,8 @@
 import java.util.List;
 import java.util.Map;
 
+import io.ably.lib.objects.LiveObjectsAdapter;
 import io.ably.lib.objects.LiveObjectsPlugin;
-import io.ably.lib.plugins.PluginConnectionAdapter;
 import io.ably.lib.rest.AblyRest;
 import io.ably.lib.rest.Auth;
 import io.ably.lib.transport.ConnectionManager;
@@ -187,9 +187,10 @@ public interface Channels extends ReadOnlyMap<String, Channel> {
     private LiveObjectsPlugin tryInitializeLiveObjectsPlugin() {
         try {
             Class<?> liveObjectsImplementation = Class.forName("io.ably.lib.objects.DefaultLiveObjectsPlugin");
+            LiveObjectsAdapter adapter = new LiveObjectsAdapter.Adapter(this);
             return (LiveObjectsPlugin) liveObjectsImplementation
-                .getDeclaredConstructor(PluginConnectionAdapter.class)
-                .newInstance(this.connection.connectionManager);
+                .getDeclaredConstructor(LiveObjectsAdapter.class)
+                .newInstance(adapter);
         } catch (ClassNotFoundException | InstantiationException | IllegalAccessException | NoSuchMethodException |
                  InvocationTargetException e) {
             Log.i(TAG, "LiveObjects plugin not found in classpath. LiveObjects functionality will not be available.", e);

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

@@ -15,7 +15,6 @@
 import io.ably.lib.debug.DebugOptions.RawProtocolListener;
 import io.ably.lib.http.HttpHelpers;
 import io.ably.lib.objects.LiveObjectsPlugin;
-import io.ably.lib.plugins.PluginConnectionAdapter;
 import io.ably.lib.realtime.AblyRealtime;
 import io.ably.lib.realtime.Channel;
 import io.ably.lib.realtime.CompletionListener;
@@ -37,7 +36,7 @@
 import io.ably.lib.util.PlatformAgentProvider;
 import io.ably.lib.util.ReconnectionStrategy;
 
-public class ConnectionManager implements ConnectListener, PluginConnectionAdapter {
+public class ConnectionManager implements ConnectListener {
     final ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
 
     /**************************************************************
@@ -1687,11 +1686,6 @@ public QueuedMessage(ProtocolMessage msg, CompletionListener listener) {
         }
     }
 
-    @Override
-    public void send(ProtocolMessage msg, CompletionListener listener) throws AblyException {
-        this.send(msg, true, listener);
-    }
-
     public void send(ProtocolMessage msg, boolean queueEvents, CompletionListener listener) throws AblyException {
         State state;
         synchronized(this) {

live-objects/src/main/kotlin/io/ably/lib/objects/DefaultLiveObjects.kt

@@ -1,8 +1,12 @@
 package io.ably.lib.objects
 
 import io.ably.lib.types.Callback
+import io.ably.lib.types.ProtocolMessage
+import io.ably.lib.util.Log
+
+internal class DefaultLiveObjects(private val adapter: LiveObjectsAdapter, private val channelName: String): LiveObjects {
+  private val tag = DefaultLiveObjects::class.simpleName
 
-internal class DefaultLiveObjects(private val channelName: String): LiveObjects {
   override fun getRoot(): LiveMap {
     TODO("Not yet implemented")
   }
@@ -43,6 +47,16 @@ internal class DefaultLiveObjects(private val channelName: String): LiveObjects
     TODO("Not yet implemented")
   }
 
+  fun handle(msg: ProtocolMessage) {
+    // RTL15b
+    msg.channelSerial?.let {
+      if (msg.action === ProtocolMessage.Action.`object`) {
+        Log.v(tag, "Setting channel serial for channelName: $channelName, value: ${msg.channelSerial}")
+        adapter.setChannelSerial(channelName, msg.channelSerial)
+      }
+    }
+  }
+
   fun dispose() {
     // Dispose of any resources associated with this LiveObjects instance
     // For example, close any open connections or clean up references

live-objects/src/main/kotlin/io/ably/lib/objects/DefaultLiveObjectsPlugin.kt

@@ -1,36 +1,19 @@
 package io.ably.lib.objects
 
-import io.ably.lib.plugins.PluginConnectionAdapter
-import io.ably.lib.realtime.CompletionListener
-import io.ably.lib.types.ErrorInfo
 import io.ably.lib.types.ProtocolMessage
-import kotlinx.coroutines.CompletableDeferred
 import java.util.concurrent.ConcurrentHashMap
 
-public class DefaultLiveObjectsPlugin(private val pluginConnectionAdapter: PluginConnectionAdapter) : LiveObjectsPlugin {
+public class DefaultLiveObjectsPlugin(private val adapter: LiveObjectsAdapter) : LiveObjectsPlugin {
 
   private val liveObjects = ConcurrentHashMap<String, DefaultLiveObjects>()
 
   override fun getInstance(channelName: String): LiveObjects {
-    return liveObjects.getOrPut(channelName) { DefaultLiveObjects(channelName) }
+    return liveObjects.getOrPut(channelName) { DefaultLiveObjects(adapter, channelName) }
   }
 
-  public suspend fun send(message: ProtocolMessage) {
-    val deferred = CompletableDeferred<Unit>()
-    pluginConnectionAdapter.send(message, object : CompletionListener {
-      override fun onSuccess() {
-        deferred.complete(Unit)
-      }
-
-      override fun onError(reason: ErrorInfo) {
-        deferred.completeExceptionally(Exception(reason.message))
-      }
-    })
-    deferred.await()
-  }
-
-  override fun handle(message: ProtocolMessage) {
-    TODO("Not yet implemented")
+  override fun handle(msg: ProtocolMessage) {
+    val channelName = msg.channel
+    liveObjects[channelName]?.handle(msg)
   }
 
   override fun dispose(channelName: String) {

live-objects/src/main/kotlin/io/ably/lib/objects/Helpers.kt

@@ -0,0 +1,24 @@
+package io.ably.lib.objects
+
+import io.ably.lib.realtime.CompletionListener
+import io.ably.lib.types.ErrorInfo
+import io.ably.lib.types.ProtocolMessage
+import kotlinx.coroutines.CompletableDeferred
+
+internal suspend fun LiveObjectsAdapter.sendAsync(message: ProtocolMessage) {
+  val deferred = CompletableDeferred<Unit>()
+  try {
+    this.send(message, object : CompletionListener {
+      override fun onSuccess() {
+        deferred.complete(Unit)
+      }
+
+      override fun onError(reason: ErrorInfo) {
+        deferred.completeExceptionally(Exception(reason.message))
+      }
+    })
+  } catch (e: Exception) {
+    deferred.completeExceptionally(e)
+  }
+  deferred.await()
+}