[ECO-5482] Implemented spec RTO15 for objects realtime publish

Author: sacOO7

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

@@ -3,7 +3,12 @@
 import io.ably.lib.realtime.AblyRealtime;
 import io.ably.lib.realtime.ChannelState;
 import io.ably.lib.realtime.CompletionListener;
-import io.ably.lib.types.*;
+import io.ably.lib.transport.ConnectionManager;
+import io.ably.lib.types.AblyException;
+import io.ably.lib.types.ChannelMode;
+import io.ably.lib.types.ChannelOptions;
+import io.ably.lib.types.ClientOptions;
+import io.ably.lib.types.ProtocolMessage;
 import io.ably.lib.util.Log;
 import org.jetbrains.annotations.NotNull;
 
@@ -67,4 +72,9 @@ public ChannelState getChannelState(@NotNull String channelName) {
     public @NotNull ClientOptions getClientOptions() {
         return ably.options;
     }
+
+    @Override
+    public @NotNull ConnectionManager getConnectionManager() {
+        return ably.connection.connectionManager;
+    }
 }

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

@@ -2,6 +2,7 @@
 
 import io.ably.lib.realtime.ChannelState;
 import io.ably.lib.realtime.CompletionListener;
+import io.ably.lib.transport.ConnectionManager;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ChannelMode;
 import io.ably.lib.types.ClientOptions;
@@ -63,5 +64,14 @@ public interface LiveObjectsAdapter {
      * @return the client options containing configuration parameters
      */
     @NotNull ClientOptions getClientOptions();
+
+    /**
+     * Retrieves the connection manager for handling connection state and operations.
+     * Used to check connection status, obtain error information, and manage
+     * message transmission across the Ably connection.
+     *
+     * @return the connection manager instance
+     */
+    @NotNull ConnectionManager getConnectionManager();
 }
 

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

@@ -106,6 +106,20 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
     objectsPool.get(ROOT_OBJECT_ID) as LiveMap
   }
 
+  /**
+   * Spec: RTO15
+   */
+  internal suspend fun publish(objectMessages: Array<ObjectMessage>) {
+    // RTO15b, RTL6c - Ensure that the channel is in a valid state for publishing
+    adapter.throwIfUnpublishableState(channelName)
+    adapter.ensureMessageSizeWithinLimit(objectMessages)
+    // RTO15e - Must construct the ProtocolMessage as per RTO15e1, RTO15e2, RTO15e3
+    val protocolMessage = ProtocolMessage(ProtocolMessage.Action.`object`, channelName)
+    protocolMessage.state = objectMessages
+    // RTO15f, RTO15g - Send the ProtocolMessage using the adapter and capture success/failure
+    adapter.sendAsync(protocolMessage)
+  }
+
   /**
    * Handles a ProtocolMessage containing proto action as `object` or `object_sync`.
    * @spec RTL1 - Processes incoming object messages and object sync messages

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

@@ -9,6 +9,9 @@ import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlin.coroutines.resume
 import kotlin.coroutines.resumeWithException
 
+/**
+ * Spec: RTO15g
+ */
 internal suspend fun LiveObjectsAdapter.sendAsync(message: ProtocolMessage) = suspendCancellableCoroutine { continuation ->
   try {
     this.send(message, object : CompletionListener {
@@ -25,11 +28,14 @@ internal suspend fun LiveObjectsAdapter.sendAsync(message: ProtocolMessage) = su
   }
 }
 
+/**
+ * Spec: RTO15d
+ */
 internal fun LiveObjectsAdapter.ensureMessageSizeWithinLimit(objectMessages: Array<ObjectMessage>) {
   val maximumAllowedSize = maxMessageSizeLimit()
   val objectsTotalMessageSize = objectMessages.sumOf { it.size() }
   if (objectsTotalMessageSize > maximumAllowedSize) {
-    throw ablyException("ObjectMessage size $objectsTotalMessageSize exceeds maximum allowed size of $maximumAllowedSize bytes",
+    throw ablyException("ObjectMessages size $objectsTotalMessageSize exceeds maximum allowed size of $maximumAllowedSize bytes",
       ErrorCode.MaxMessageSizeExceeded)
   }
 }
@@ -53,6 +59,13 @@ internal fun LiveObjectsAdapter.throwIfInvalidWriteApiConfiguration(channelName:
   throwIfInChannelState(channelName, arrayOf(ChannelState.detached, ChannelState.failed, ChannelState.suspended))
 }
 
+internal fun LiveObjectsAdapter.throwIfUnpublishableState(channelName: String) {
+  if (!connectionManager.isActive) {
+    throw ablyException(connectionManager.stateErrorInfo)
+  }
+  throwIfInChannelState(channelName, arrayOf(ChannelState.failed, ChannelState.suspended))
+}
+
 // Spec: RTO2
 internal fun LiveObjectsAdapter.throwIfMissingChannelMode(channelName: String, channelMode: ChannelMode) {
   val channelModes = getChannelModes(channelName)