1. Added coroutinex as a runtime and test dependency to liveobjects

2. Added bridging interfaces to send and receive protocol messages

Author: sacOO7

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

@@ -1,16 +1,19 @@
 package io.ably.lib.objects;
 
+import io.ably.lib.plugins.PluginInstance;
 import org.jetbrains.annotations.NotNull;
 
 /**
- * The LiveObjectsPlugin interface provides a mechanism to retrieve instances of LiveObjects
- * associated with specific channel names. This allows for interaction with live data objects
- * in a real-time environment.
+ * The LiveObjectsPlugin interface provides a mechanism for managing and interacting with
+ * live data objects in a real-time environment. It allows for the retrieval, disposal, and
+ * management of LiveObjects instances associated with specific channel names.
  */
-public interface LiveObjectsPlugin {
+public interface LiveObjectsPlugin extends PluginInstance {
 
     /**
      * Retrieves an instance of LiveObjects associated with the specified channel name.
+     * This method ensures that a LiveObjects instance is available for the given channel,
+     * creating one if it does not already exist.
      *
      * @param channelName the name of the channel for which the LiveObjects instance is to be retrieved.
      * @return the LiveObjects instance associated with the specified channel name.
@@ -19,6 +22,8 @@ public interface LiveObjectsPlugin {
 
     /**
      * Disposes of the LiveObjects instance associated with the specified channel name.
+     * This method removes the LiveObjects instance for the given channel, releasing any
+     * resources associated with it.
      *
      * @param channelName the name of the channel whose LiveObjects instance is to be removed.
      */

lib/src/main/java/io/ably/lib/plugins/PluginConnectionAdapter.java

@@ -0,0 +1,25 @@
+package io.ably.lib.plugins;
+
+import io.ably.lib.realtime.CompletionListener;
+import io.ably.lib.types.AblyException;
+import io.ably.lib.types.ProtocolMessage;
+
+/**
+ * The PluginConnectionAdapter interface defines a contract for managing real-time communication
+ * between plugins and the Ably Realtime system. Implementations of this interface are responsible
+ * for sending protocol messages to their intended recipients, optionally queuing events, and
+ * notifying listeners of the operation's outcome.
+ */
+public interface PluginConnectionAdapter {
+
+    /**
+     * Sends a protocol message to its intended recipient.
+     * This method transmits a protocol message, allowing for queuing events if necessary,
+     * and notifies the provided listener upon the success or failure of the send operation.
+     *
+     * @param msg the protocol message to send.
+     * @param listener a listener to be notified of the success or failure of the send operation.
+     * @throws AblyException if an error occurs during the send operation.
+     */
+    void send(ProtocolMessage msg, CompletionListener listener) throws AblyException;
+}

lib/src/main/java/io/ably/lib/plugins/PluginInstance.java

@@ -0,0 +1,25 @@
+package io.ably.lib.plugins;
+
+import io.ably.lib.types.ProtocolMessage;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * The ProtocolMessageHandler interface defines a contract for handling protocol messages.
+ * Implementations of this interface are responsible for processing incoming protocol messages
+ * and performing the necessary actions based on the message content.
+ */
+public interface PluginInstance {
+    /**
+     * Handles a protocol message.
+     * This method is invoked whenever a protocol message is received, allowing the implementation
+     * to process the message and take appropriate actions.
+     *
+     * @param message the protocol message to handle.
+     */
+    void handle(@NotNull ProtocolMessage message);
+
+    /**
+     * Disposes of the plugin instance and all underlying resources.
+     */
+    void dispose();
+}

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

@@ -7,6 +7,7 @@
 import java.util.Map;
 
 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;
@@ -71,7 +72,10 @@ public AblyRealtime(ClientOptions options) throws AblyException {
         super(options);
         final InternalChannels channels = new InternalChannels();
         this.channels = channels;
-        connection = new Connection(this, channels, platformAgentProvider);
+
+        liveObjectsPlugin = tryInitializeLiveObjectsPlugin();
+
+        connection = new Connection(this, channels, platformAgentProvider, liveObjectsPlugin);
 
         if (!StringUtils.isNullOrEmpty(options.recover)) {
             RecoveryKeyContext recoveryKeyContext = RecoveryKeyContext.decode(options.recover);
@@ -81,8 +85,6 @@ public AblyRealtime(ClientOptions options) throws AblyException {
             }
         }
 
-        liveObjectsPlugin = tryInitializeLiveObjectsPlugin();
-
         if(options.autoConnect) connection.connect();
     }
 
@@ -182,7 +184,9 @@ public interface Channels extends ReadOnlyMap<String, Channel> {
     private LiveObjectsPlugin tryInitializeLiveObjectsPlugin() {
         try {
             Class<?> liveObjectsImplementation = Class.forName("io.ably.lib.objects.DefaultLiveObjectsPlugin");
-            return (LiveObjectsPlugin) liveObjectsImplementation.getDeclaredConstructor().newInstance();
+            return (LiveObjectsPlugin) liveObjectsImplementation
+                .getDeclaredConstructor(PluginConnectionAdapter.class)
+                .newInstance(this.connection.connectionManager);
         } catch (ClassNotFoundException | InstantiationException | IllegalAccessException | NoSuchMethodException |
                  InvocationTargetException e) {
             Log.w(TAG, "LiveObjects plugin not found in classpath. LiveObjects functionality will not be available.", e);

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

@@ -1,5 +1,6 @@
 package io.ably.lib.realtime;
 
+import io.ably.lib.objects.LiveObjectsPlugin;
 import io.ably.lib.realtime.ConnectionStateListener.ConnectionStateChange;
 import io.ably.lib.transport.ConnectionManager;
 import io.ably.lib.types.AblyException;
@@ -122,10 +123,10 @@ public void close() {
      * internal
      *****************/
 
-    Connection(AblyRealtime ably, ConnectionManager.Channels channels, PlatformAgentProvider platformAgentProvider) throws AblyException {
+    Connection(AblyRealtime ably, ConnectionManager.Channels channels, PlatformAgentProvider platformAgentProvider, LiveObjectsPlugin liveObjectsPlugin) throws AblyException {
         this.ably = ably;
         this.state = ConnectionState.initialized;
-        this.connectionManager = new ConnectionManager(ably, this, channels, platformAgentProvider);
+        this.connectionManager = new ConnectionManager(ably, this, channels, platformAgentProvider, liveObjectsPlugin);
     }
 
     public void onConnectionStateChange(ConnectionStateChange stateChange) {

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

@@ -14,6 +14,8 @@
 import io.ably.lib.debug.DebugOptions;
 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;
@@ -35,7 +37,7 @@
 import io.ably.lib.util.PlatformAgentProvider;
 import io.ably.lib.util.ReconnectionStrategy;
 
-public class ConnectionManager implements ConnectListener {
+public class ConnectionManager implements ConnectListener, PluginConnectionAdapter {
     final ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
 
     /**************************************************************
@@ -79,6 +81,13 @@ public class ConnectionManager implements ConnectListener {
      */
     private boolean cleaningUpAfterEnteringTerminalState = false;
 
+    /**
+     * A nullable reference to the LiveObjects plugin.
+     * <p>
+     * This field is initialized only if the LiveObjects plugin is present in the classpath.
+     */
+    private final LiveObjectsPlugin liveObjectsPlugin;
+
     /**
      * Methods on the channels map owned by the {@link AblyRealtime} instance
      * which the {@link ConnectionManager} needs access to.
@@ -764,11 +773,12 @@ public void run() {
      * ConnectionManager
      ***********************/
 
-    public ConnectionManager(final AblyRealtime ably, final Connection connection, final Channels channels, final PlatformAgentProvider platformAgentProvider) throws AblyException {
+    public ConnectionManager(final AblyRealtime ably, final Connection connection, final Channels channels, final PlatformAgentProvider platformAgentProvider, LiveObjectsPlugin liveObjectsPlugin) throws AblyException {
         this.ably = ably;
         this.connection = connection;
         this.channels = channels;
         this.platformAgentProvider = platformAgentProvider;
+        this.liveObjectsPlugin = liveObjectsPlugin;
 
         ClientOptions options = ably.options;
         this.hosts = new Hosts(options.realtimeHost, Defaults.HOST_REALTIME, options);
@@ -1220,6 +1230,12 @@ public void onMessage(ITransport transport, ProtocolMessage message) throws Ably
                 case auth:
                     addAction(new ReauthAction());
                     break;
+                case object:
+                case object_sync:
+                    if (liveObjectsPlugin != null) {
+                        liveObjectsPlugin.handle(message);
+                    }
+                    break;
                 default:
                     onChannelMessage(message);
             }
@@ -1667,6 +1683,11 @@ 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) {

lib/src/test/java/io/ably/lib/test/realtime/ConnectionManagerTest.java

@@ -137,7 +137,7 @@ public void connectionmanager_fallback_none_withoutconnection() throws AblyExcep
             Connection connection = Mockito.mock(Connection.class);
             final ConnectionManager.Channels channels = Mockito.mock(ConnectionManager.Channels.class);
 
-            ConnectionManager connectionManager = new ConnectionManager(ably, connection, channels, new EmptyPlatformAgentProvider()) {
+            ConnectionManager connectionManager = new ConnectionManager(ably, connection, channels, new EmptyPlatformAgentProvider(), null) {
                 @Override
                 protected boolean checkConnectivity() {
                     return false;

live-objects/build.gradle.kts

@@ -10,6 +10,9 @@ repositories {
 dependencies {
     implementation(project(":java"))
     testImplementation(kotlin("test"))
+    implementation(libs.coroutine.core)
+
+    testImplementation(libs.coroutine.test)
 }
 
 tasks.test {

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

@@ -1,10 +1,15 @@
 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 : LiveObjectsPlugin {
+public class DefaultLiveObjectsPlugin(private val pluginConnectionAdapter: PluginConnectionAdapter) : LiveObjectsPlugin {
 
-  private val liveObjects = ConcurrentHashMap<String, LiveObjects>()
+  private val liveObjects = ConcurrentHashMap<String, DefaultLiveObjects>()
 
   override fun getInstance(channelName: String): LiveObjects {
     return liveObjects.getOrPut(channelName) { DefaultLiveObjects(channelName) }
@@ -13,4 +18,26 @@ public class DefaultLiveObjectsPlugin : LiveObjectsPlugin {
   override fun dispose(channelName: String) {
     liveObjects.remove(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 dispose() {
+    TODO("Not yet implemented")
+  }
 }