[ECO-5457] Moved callbackScope at global level to be shared amongst all public api methods

Author: sacOO7

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

@@ -46,13 +46,15 @@ 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.
+     * This is invoked when ablyRealtimeClient.channels.release(channelName) is called
      *
      * @param channelName the name of the channel whose LiveObjects instance is to be removed.
      */
     void dispose(@NotNull String channelName);
 
     /**
      * Disposes of the plugin instance and all underlying resources.
+     * This is invoked when ablyRealtimeClient.close() is called
      */
     void dispose();
 }

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

@@ -9,6 +9,7 @@ import io.ably.lib.util.Log
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import kotlinx.coroutines.flow.MutableSharedFlow
+import java.util.concurrent.CancellationException
 
 /**
  * Default implementation of LiveObjects interface.
@@ -34,12 +35,6 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
   private val sequentialScope =
     CoroutineScope(Dispatchers.Default.limitedParallelism(1) + CoroutineName(channelName) + SupervisorJob())
 
-  /**
-   * Coroutine scope for handling callbacks asynchronously.
-   */
-  private val callbackScope =
-    CoroutineScope(Dispatchers.Default + CoroutineName("LiveObjectsCallback-$channelName") + SupervisorJob())
-
   /**
    * Event bus for handling incoming object messages sequentially.
    */
@@ -50,12 +45,10 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
     incomingObjectsHandler = initializeHandlerForIncomingObjectMessages()
   }
 
-  override fun getRoot(): LiveMap {
-    return runBlocking { getRootAsync() }
-  }
+  override fun getRoot(): LiveMap = runBlocking { getRootAsync() }
 
   override fun getRootAsync(callback: Callback<LiveMap>) {
-    callbackScope.launchWithCallback(callback) { getRootAsync() }
+    GlobalCallbackScope.launchWithCallback(callback) { getRootAsync() }
   }
 
   override fun createMap(liveMap: LiveMap): LiveMap {
@@ -97,12 +90,10 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
 
   override fun offAll() = objectsManager.offAll()
 
-  private suspend fun getRootAsync(): LiveMap {
-    return sequentialScope.async {
-      adapter.throwIfInvalidAccessApiConfiguration(channelName)
-      objectsManager.ensureSynced(state)
-      objectsPool.get(ROOT_OBJECT_ID) as LiveMap
-    }.await()
+  private suspend fun getRootAsync(): LiveMap = withContext(sequentialScope.coroutineContext) {
+    adapter.throwIfInvalidAccessApiConfiguration(channelName)
+    objectsManager.ensureSynced(state)
+    objectsPool.get(ROOT_OBJECT_ID) as LiveMap
   }
 
   /**
@@ -188,9 +179,12 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
   }
 
   // Dispose of any resources associated with this LiveObjects instance
-  fun dispose() {
-    incomingObjectsHandler.cancel() // objectsEventBus automatically garbage collected when collector is cancelled
+  fun dispose(reason: String) {
+    val cancellationError = CancellationException("Objects disposed for channel $channelName, reason: $reason")
+    incomingObjectsHandler.cancel(cancellationError) // objectsEventBus automatically garbage collected when collector is cancelled
     objectsPool.dispose()
     objectsManager.dispose()
+    // Don't cancel sequentialScope (needed in public methods), just cancel ongoing coroutines
+    sequentialScope.coroutineContext.cancelChildren(cancellationError)
   }
 }

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

@@ -22,13 +22,13 @@ public class DefaultLiveObjectsPlugin(private val adapter: LiveObjectsAdapter) :
   }
 
   override fun dispose(channelName: String) {
-    liveObjects[channelName]?.dispose()
+    liveObjects[channelName]?.dispose("Channel has ben released using channels.release()")
     liveObjects.remove(channelName)
   }
 
   override fun dispose() {
     liveObjects.values.forEach {
-      it.dispose()
+      it.dispose("AblyClient has been closed using client.close()")
     }
     liveObjects.clear()
   }

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

@@ -49,26 +49,26 @@ internal val String.byteSize: Int
   get() = this.toByteArray(Charsets.UTF_8).size
 
 /**
- * Executes a suspend function within a coroutine and handles the result via a callback.
- *
- * This utility bridges between coroutine-based implementation code and callback-based APIs.
- * It launches a coroutine in the current scope to execute the provided suspend block,
- * then routes the result or any error to the appropriate callback method.
- *
- * @param T The type of result expected from the operation
- * @param callback The callback to invoke with the operation result or error
- * @param block The suspend function to execute that returns a value of type T
+ * A global coroutine scope for executing callbacks asynchronously.
+ * Provides safe execution of suspend functions with results delivered via callbacks,
+ * with proper error handling for both the execution and callback invocation.
  */
-internal fun <T> CoroutineScope.launchWithCallback(callback: Callback<T>, block: suspend () -> T) {
-  launch {
-    try {
-      val result = block()
-      try { callback.onSuccess(result) } catch (t: Throwable) {
-        Log.e("asyncCallback", "Error occurred while executing callback's onSuccess handler", t)
-      } // catch and don't rethrow error from callback
-    } catch (throwable: Throwable) {
-      val exception = throwable as? AblyException
-      callback.onError(exception?.errorInfo)
+internal object GlobalCallbackScope {
+  private const val TAG = "GlobalCallbackScope"
+  private val scope =
+    CoroutineScope(Dispatchers.Default + CoroutineName("LiveObjects-GlobalCallbackScope") + SupervisorJob())
+
+  internal fun <T> launchWithCallback(callback: Callback<T>, block: suspend () -> T) {
+    scope.launch {
+      try {
+        val result = block()
+        try { callback.onSuccess(result) } catch (t: Throwable) {
+          Log.e(TAG, "Error occurred while executing callback's onSuccess handler", t)
+        } // catch and don't rethrow error from callback
+      } catch (throwable: Throwable) {
+        val exception = throwable as? AblyException
+        callback.onError(exception?.errorInfo)
+      }
     }
   }
 }

live-objects/src/test/kotlin/io/ably/lib/objects/unit/objects/DefaultLiveObjectsTest.kt

@@ -57,15 +57,16 @@ class DefaultLiveObjectsTest {
     // RTO4b - If the HAS_OBJECTS flag is 0, the sync sequence must be considered complete immediately
     defaultLiveObjects.handleStateChange(ChannelState.attached, false)
 
+    // Verify expected outcomes
+    assertWaiter { defaultLiveObjects.state == ObjectsState.SYNCED } // RTO4b4
+
     verify(exactly = 1) {
       defaultLiveObjects.objectsPool.resetToInitialPool(true)
     }
     verify(exactly = 1) {
       defaultLiveObjects.ObjectsManager.endSync(any<Boolean>())
     }
 
-    // Verify expected outcomes
-    assertWaiter { defaultLiveObjects.state == ObjectsState.SYNCED } // RTO4b4
     assertEquals(0, defaultLiveObjects.ObjectsManager.SyncObjectsDataPool.size) // RTO4b3
     assertEquals(0, defaultLiveObjects.ObjectsManager.BufferedObjectOperations.size) // RTO4b5
     assertEquals(1, defaultLiveObjects.objectsPool.size()) // RTO4b1 - Only root remains