Merge branch 'feature/object-subscriptions' into feature/server-provided-tombstone-serial

Author: sacOO7

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

@@ -5,6 +5,7 @@ import io.ably.lib.objects.state.ObjectsStateEvent
 import io.ably.lib.objects.type.counter.LiveCounter
 import io.ably.lib.objects.type.map.LiveMap
 import io.ably.lib.realtime.ChannelState
+import io.ably.lib.types.AblyException
 import io.ably.lib.types.Callback
 import io.ably.lib.types.ProtocolMessage
 import io.ably.lib.util.Log
@@ -24,7 +25,7 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
    */
   internal val objectsPool = ObjectsPool(this)
 
-  internal var state = ObjectsState.INITIALIZED
+  internal var state = ObjectsState.Initialized
 
   /**
    * @spec RTO4 - Used for handling object messages and object sync messages
@@ -37,6 +38,11 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
   private val sequentialScope =
     CoroutineScope(Dispatchers.Default.limitedParallelism(1) + CoroutineName(channelName) + SupervisorJob())
 
+  /**
+   * Provides a channel-specific scope for safely executing asynchronous operations with callbacks.
+   */
+  private val callbackScope = ObjectsCallbackScope(channelName)
+
   /**
    * Event bus for handling incoming object messages sequentially.
    */
@@ -49,10 +55,6 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
 
   override fun getRoot(): LiveMap = runBlocking { getRootAsync() }
 
-  override fun getRootAsync(callback: Callback<LiveMap>) {
-    GlobalCallbackScope.launchWithCallback(callback) { getRootAsync() }
-  }
-
   override fun createMap(liveMap: LiveMap): LiveMap {
     TODO("Not yet implemented")
   }
@@ -65,6 +67,10 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
     TODO("Not yet implemented")
   }
 
+  override fun getRootAsync(callback: Callback<LiveMap>) {
+    callbackScope.launchWithCallback(callback) { getRootAsync() }
+  }
+
   override fun createMapAsync(liveMap: LiveMap, callback: Callback<LiveMap>) {
     TODO("Not yet implemented")
   }
@@ -157,7 +163,7 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
           Log.v(tag, "Objects.onAttached() channel=$channelName, hasObjects=$hasObjects")
 
           // RTO4a
-          val fromInitializedState = this@DefaultLiveObjects.state == ObjectsState.INITIALIZED
+          val fromInitializedState = this@DefaultLiveObjects.state == ObjectsState.Initialized
           if (hasObjects || fromInitializedState) {
             // should always start a new sync sequence if we're in the initialized state, no matter the HAS_OBJECTS flag value.
             // this guarantees we emit both "syncing" -> "synced" events in that order.
@@ -191,12 +197,13 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
   }
 
   // Dispose of any resources associated with this LiveObjects instance
-  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
+  fun dispose(cause: AblyException) {
+    val disposeReason = CancellationException().apply { initCause(cause) }
+    incomingObjectsHandler.cancel(disposeReason) // 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)
+    // Don't cancel sequentialScope (needed in getRoot method), just cancel ongoing coroutines
+    sequentialScope.coroutineContext.cancelChildren(disposeReason)
+    callbackScope.cancel(disposeReason) // cancel all ongoing callbacks
   }
 }

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("Channel has ben released using channels.release()")
+    liveObjects[channelName]?.dispose(clientError("Channel has ben released using channels.release()"))
     liveObjects.remove(channelName)
   }
 
   override fun dispose() {
     liveObjects.values.forEach {
-      it.dispose("AblyClient has been closed using client.close()")
+      it.dispose(clientError("AblyClient has been closed using client.close()"))
     }
     liveObjects.clear()
   }

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

@@ -28,7 +28,7 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects): Obje
    * @spec RTO8 - Buffers messages if not synced, applies immediately if synced
    */
   internal fun handleObjectMessages(objectMessages: List<ObjectMessage>) {
-    if (liveObjects.state != ObjectsState.SYNCED) {
+    if (liveObjects.state != ObjectsState.Synced) {
       // RTO7 - The client receives object messages in realtime over the channel concurrently with the sync sequence.
       // Some of the incoming object messages may have already been applied to the objects described in
       // the sync sequence, but others may not; therefore we must buffer these messages so that we can apply
@@ -78,7 +78,7 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects): Obje
     bufferedObjectOperations.clear() // RTO5a2b
     syncObjectsDataPool.clear() // RTO5a2a
     currentSyncId = syncId
-    stateChange(ObjectsState.SYNCING, false)
+    stateChange(ObjectsState.Syncing, false)
   }
 
   /**
@@ -96,7 +96,7 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects): Obje
     bufferedObjectOperations.clear() // RTO5c5
     syncObjectsDataPool.clear() // RTO5c4
     currentSyncId = null // RTO5c3
-    stateChange(ObjectsState.SYNCED, deferStateEvent)
+    stateChange(ObjectsState.Synced, deferStateEvent)
   }
 
   /**

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

@@ -10,9 +10,9 @@ import kotlinx.coroutines.*
  * @spec RTO2 - enum representing objects state
  */
 internal enum class ObjectsState {
-  INITIALIZED,
-  SYNCING,
-  SYNCED
+  Initialized,
+  Syncing,
+  Synced
 }
 
 /**
@@ -21,9 +21,9 @@ internal enum class ObjectsState {
  * INITIALIZED maps to null (no event), while SYNCING and SYNCED map to their respective events.
  */
 private val objectsStateToEventMap = mapOf(
-  ObjectsState.INITIALIZED to null,
-  ObjectsState.SYNCING to ObjectsStateEvent.SYNCING,
-  ObjectsState.SYNCED to ObjectsStateEvent.SYNCED
+  ObjectsState.Initialized to null,
+  ObjectsState.Syncing to ObjectsStateEvent.SYNCING,
+  ObjectsState.Synced to ObjectsStateEvent.SYNCED
 )
 
 /**
@@ -63,8 +63,6 @@ internal abstract class ObjectsStateCoordinator : ObjectsStateChange, HandlesObj
   // related to RTC10, should have a separate EventEmitter for users of the library
   private val externalObjectStateEmitter = ObjectsStateEmitter()
 
-  private val emitterScope = CoroutineScope(Dispatchers.Default.limitedParallelism(1) + SupervisorJob())
-
   override fun on(event: ObjectsStateEvent, listener: ObjectsStateChange.Listener): ObjectsSubscription {
     externalObjectStateEmitter.on(event, listener)
     return ObjectsSubscription {
@@ -78,16 +76,13 @@ internal abstract class ObjectsStateCoordinator : ObjectsStateChange, HandlesObj
 
   override fun objectsStateChanged(newState: ObjectsState) {
     objectsStateToEventMap[newState]?.let { objectsStateEvent ->
-      // emitterScope makes sure next launch can only start when previous launch finishes
-      emitterScope.launch {
-        internalObjectStateEmitter.emit(objectsStateEvent)
-        externalObjectStateEmitter.emit(objectsStateEvent)
-      }
+      internalObjectStateEmitter.emit(objectsStateEvent)
+      externalObjectStateEmitter.emit(objectsStateEvent)
     }
   }
 
   override suspend fun ensureSynced(currentState: ObjectsState) {
-    if (currentState != ObjectsState.SYNCED) {
+    if (currentState != ObjectsState.Synced) {
       val deferred = CompletableDeferred<Unit>()
       internalObjectStateEmitter.once(ObjectsStateEvent.SYNCED) {
         Log.v(tag, "Objects state changed to SYNCED, resuming ensureSynced")
@@ -97,10 +92,7 @@ internal abstract class ObjectsStateCoordinator : ObjectsStateChange, HandlesObj
     }
   }
 
-  override fun disposeObjectsStateListeners() {
-    offAll()
-    emitterScope.cancel("ObjectsManager disposed")
-  }
+  override fun disposeObjectsStateListeners() = offAll()
 }
 
 private class ObjectsStateEmitter : EventEmitter<ObjectsStateEvent, ObjectsStateChange.Listener>() {

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

@@ -5,6 +5,7 @@ import io.ably.lib.types.Callback
 import io.ably.lib.types.ErrorInfo
 import io.ably.lib.util.Log
 import kotlinx.coroutines.*
+import java.util.concurrent.CancellationException
 
 internal fun ablyException(
   errorMessage: String,
@@ -49,26 +50,31 @@ internal val String.byteSize: Int
   get() = this.toByteArray(Charsets.UTF_8).size
 
 /**
- * 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.
+ * A channel-specific coroutine scope for executing callbacks asynchronously in the LiveObjects system.
+ * Provides safe execution of suspend functions with results delivered via callbacks.
+ * Supports proper error handling and cancellation during LiveObjects disposal.
  */
-internal object GlobalCallbackScope {
-  private const val TAG = "GlobalCallbackScope"
+internal class ObjectsCallbackScope(channelName: String) {
+  private val tag = "ObjectsCallbackScope-$channelName"
+
   private val scope =
-    CoroutineScope(Dispatchers.Default + CoroutineName("LiveObjects-GlobalCallbackScope") + SupervisorJob())
+    CoroutineScope(Dispatchers.Default + CoroutineName(tag) + 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)
+          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)
       }
     }
   }
+
+  internal fun cancel(cause: CancellationException) {
+    scope.coroutineContext.cancelChildren(cause)
+  }
 }

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

@@ -52,11 +52,7 @@ internal class ObjectDataJsonSerializer : JsonSerializer<ObjectData>, JsonDeseri
         is String -> obj.addProperty("string", v)
         is Number -> obj.addProperty("number", v.toDouble())
         is Binary -> obj.addProperty("bytes", Base64.getEncoder().encodeToString(v.data))
-        // Spec: OD4c5
-        is JsonObject, is JsonArray -> {
-          obj.addProperty("string", v.toString())
-          obj.addProperty("encoding", "json")
-        }
+        is JsonObject, is JsonArray -> obj.addProperty("json", v.toString()) // Spec: OD4c5
       }
     }
     return obj
@@ -65,24 +61,12 @@ internal class ObjectDataJsonSerializer : JsonSerializer<ObjectData>, JsonDeseri
   override fun deserialize(json: JsonElement, typeOfT: Type?, context: JsonDeserializationContext?): ObjectData {
     val obj = if (json.isJsonObject) json.asJsonObject else throw JsonParseException("Expected JsonObject")
     val objectId = if (obj.has("objectId")) obj.get("objectId").asString else null
-    val encoding = if (obj.has("encoding")) obj.get("encoding").asString else null
     val value = when {
       obj.has("boolean") -> ObjectValue(obj.get("boolean").asBoolean)
-      // Spec: OD5b3
-      obj.has("string") && encoding == "json" -> {
-        val jsonStr = obj.get("string").asString
-        val parsed = JsonParser.parseString(jsonStr)
-        ObjectValue(
-          when {
-            parsed.isJsonObject -> parsed.asJsonObject
-            parsed.isJsonArray -> parsed.asJsonArray
-            else -> throw JsonParseException("Invalid JSON string for encoding=json")
-          }
-        )
-      }
       obj.has("string") -> ObjectValue(obj.get("string").asString)
       obj.has("number") -> ObjectValue(obj.get("number").asDouble)
       obj.has("bytes") -> ObjectValue(Binary(Base64.getDecoder().decode(obj.get("bytes").asString)))
+      obj.has("json") -> ObjectValue(JsonParser.parseString(obj.get("json").asString))
       else -> {
         if (objectId != null)
           null

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

@@ -1,7 +1,6 @@
 package io.ably.lib.objects.serialization
 
 import com.google.gson.JsonArray
-import com.google.gson.JsonElement
 import com.google.gson.JsonObject
 import com.google.gson.JsonParser
 import io.ably.lib.objects.*
@@ -634,9 +633,6 @@ private fun ObjectData.writeMsgpack(packer: MessagePacker) {
   if (objectId != null) fieldCount++
   value?.let {
     fieldCount++
-    if (it.value is JsonElement) {
-      fieldCount += 1 // For extra "encoding" field
-    }
   }
 
   packer.packMapHeader(fieldCount)
@@ -666,10 +662,8 @@ private fun ObjectData.writeMsgpack(packer: MessagePacker) {
         packer.writePayload(v.data)
       }
       is JsonObject, is JsonArray -> {
-        packer.packString("string")
-        packer.packString(v.toString())
-        packer.packString("encoding")
         packer.packString("json")
+        packer.packString(v.toString())
       }
     }
   }
@@ -682,8 +676,6 @@ private fun readObjectData(unpacker: MessageUnpacker): ObjectData {
   val fieldCount = unpacker.unpackMapHeader()
   var objectId: String? = null
   var value: ObjectValue? = null
-  var encoding: String? = null
-  var stringValue: String? = null
 
   for (i in 0 until fieldCount) {
     val fieldName = unpacker.unpackString().intern()
@@ -697,33 +689,29 @@ private fun readObjectData(unpacker: MessageUnpacker): ObjectData {
     when (fieldName) {
       "objectId" -> objectId = unpacker.unpackString()
       "boolean" -> value = ObjectValue(unpacker.unpackBoolean())
-      "string" -> stringValue = unpacker.unpackString()
+      "string" -> value = ObjectValue(unpacker.unpackString())
       "number" -> value = ObjectValue(unpacker.unpackDouble())
       "bytes" -> {
         val size = unpacker.unpackBinaryHeader()
         val bytes = ByteArray(size)
         unpacker.readPayload(bytes)
         value = ObjectValue(Binary(bytes))
       }
-      "encoding" -> encoding = unpacker.unpackString()
+      "json" -> {
+        val jsonString = unpacker.unpackString()
+        val parsed = JsonParser.parseString(jsonString)
+        value = ObjectValue(
+          when {
+            parsed.isJsonObject -> parsed.asJsonObject
+            parsed.isJsonArray -> parsed.asJsonArray
+            else ->
+              throw ablyException("Invalid JSON string for json field", ErrorCode.MapValueDataTypeUnsupported, HttpStatusCode.InternalServerError)
+          }
+        )
+      }
       else -> unpacker.skipValue()
     }
   }
 
-  // Handle string with encoding if needed
-  if (stringValue != null && encoding == "json") {
-    val parsed = JsonParser.parseString(stringValue)
-    value = ObjectValue(
-      when {
-        parsed.isJsonObject -> parsed.asJsonObject
-        parsed.isJsonArray -> parsed.asJsonArray
-        else ->
-          throw ablyException("Invalid JSON string for encoding=json", ErrorCode.MapValueDataTypeUnsupported, HttpStatusCode.InternalServerError)
-      }
-    )
-  } else if (stringValue != null) {
-    value = ObjectValue(stringValue)
-  }
-
   return ObjectData(objectId = objectId, value = value)
 }

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

@@ -38,7 +38,7 @@ class DefaultLiveObjectsTest : IntegrationTest() {
     val objects = channel.objects
     assertNotNull(objects)
 
-    assertEquals(ObjectsState.INITIALIZED, objects.State, "Initial state should be INITIALIZED")
+    assertEquals(ObjectsState.Initialized, objects.State, "Initial state should be INITIALIZED")
 
     val syncStates = mutableListOf<ObjectsStateEvent>()
     objects.on(ObjectsStateEvent.SYNCING) {

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

@@ -34,7 +34,7 @@ class DefaultLiveObjectsTest {
     // RTO4a - If the HAS_OBJECTS flag is 1, the server will shortly perform an OBJECT_SYNC sequence
     defaultLiveObjects.handleStateChange(ChannelState.attached, true)
 
-    assertWaiter { defaultLiveObjects.state == ObjectsState.SYNCING }
+    assertWaiter { defaultLiveObjects.state == ObjectsState.Syncing }
 
     // It is expected that the client will start a new sync sequence
     verify(exactly = 1) {
@@ -59,7 +59,7 @@ class DefaultLiveObjectsTest {
     defaultLiveObjects.handleStateChange(ChannelState.attached, false)
 
     // Verify expected outcomes
-    assertWaiter { defaultLiveObjects.state == ObjectsState.SYNCED } // RTO4b4
+    assertWaiter { defaultLiveObjects.state == ObjectsState.Synced } // RTO4b4
 
     verify(exactly = 1) {
       defaultLiveObjects.objectsPool.resetToInitialPool(true)
@@ -80,7 +80,7 @@ class DefaultLiveObjectsTest {
     val defaultLiveObjects = getDefaultLiveObjectsWithMockedDeps()
 
     // Ensure we're in INITIALIZED state
-    defaultLiveObjects.state = ObjectsState.INITIALIZED
+    defaultLiveObjects.state = ObjectsState.Initialized
 
     // RTO4a - Should start sync even with HAS_OBJECTS flag false when in INITIALIZED state
     defaultLiveObjects.handleStateChange(ChannelState.attached, false)