[ECO-5426] Updated code, marked data types in objectsPool, LiveCounter and LiveMap as thread safe

Author: sacOO7

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

@@ -21,12 +21,12 @@ internal enum class ObjectsState {
  * Default implementation of LiveObjects interface.
  * Provides the core functionality for managing live objects on a channel.
  */
-internal class DefaultLiveObjects(private val channelName: String, internal val adapter: LiveObjectsAdapter): LiveObjects {
+internal class DefaultLiveObjects(internal val channelName: String, internal val adapter: LiveObjectsAdapter): LiveObjects {
   private val tag = "DefaultLiveObjects"
   /**
    * @spec RTO3 - Objects pool storing all live objects by object ID
    */
-  internal val objectsPool = ObjectsPool(adapter)
+  internal val objectsPool = ObjectsPool(this)
 
   internal var state = ObjectsState.INITIALIZED
 
@@ -203,9 +203,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/ObjectsManager.kt

@@ -214,8 +214,8 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
    */
   private fun createObjectFromState(objectState: ObjectState): BaseLiveObject {
     return when {
-      objectState.counter != null -> DefaultLiveCounter.zeroValue(objectState.objectId, liveObjects.adapter) // RTO5c1b1a
-      objectState.map != null -> DefaultLiveMap.zeroValue(objectState.objectId, liveObjects.adapter, liveObjects.objectsPool) // RTO5c1b1b
+      objectState.counter != null -> DefaultLiveCounter.zeroValue(objectState.objectId, liveObjects) // RTO5c1b1a
+      objectState.map != null -> DefaultLiveMap.zeroValue(objectState.objectId, liveObjects) // RTO5c1b1b
       else -> throw clientError("Object state must contain either counter or map data") // RTO5c1b1c
     }
   }

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

@@ -6,6 +6,7 @@ import io.ably.lib.objects.type.livecounter.DefaultLiveCounter
 import io.ably.lib.objects.type.livemap.DefaultLiveMap
 import io.ably.lib.util.Log
 import kotlinx.coroutines.*
+import java.util.concurrent.ConcurrentHashMap
 
 /**
  * Constants for ObjectsPool configuration
@@ -32,14 +33,15 @@ internal const val ROOT_OBJECT_ID = "root"
  * @spec RTO3 - Maintains an objects pool for all live objects on the channel
  */
 internal class ObjectsPool(
-  private val adapter: LiveObjectsAdapter
+  private val liveObjects: DefaultLiveObjects
 ) {
   private val tag = "ObjectsPool"
 
   /**
+   * ConcurrentHashMap for thread-safe access from public APIs in LiveMap and LiveCounter.
    * @spec RTO3a - Pool storing all live objects by object ID
    */
-  private val pool = mutableMapOf<String, BaseLiveObject>()
+  private val pool = ConcurrentHashMap<String, BaseLiveObject>()
 
   /**
    * Coroutine scope for garbage collection
@@ -48,22 +50,12 @@ internal class ObjectsPool(
   private var gcJob: Job // Job for the garbage collection coroutine
 
   init {
-    // Initialize pool with root object
-    createInitialPool()
+    // RTO3b - Initialize pool with root object
+    pool[ROOT_OBJECT_ID] = DefaultLiveMap.zeroValue(ROOT_OBJECT_ID, liveObjects)
     // Start garbage collection coroutine
     gcJob = startGCJob()
   }
 
-  /**
-   * Creates the initial pool with root object.
-   *
-   * @spec RTO3b - Creates root LiveMap object
-   */
-  private fun createInitialPool() {
-    val root = DefaultLiveMap.zeroValue(ROOT_OBJECT_ID, adapter, this)
-    pool[ROOT_OBJECT_ID] = root
-  }
-
   /**
    * Gets a live object from the pool by object ID.
    */
@@ -119,8 +111,8 @@ internal class ObjectsPool(
 
     val parsedObjectId = ObjectId.fromString(objectId) // RTO6b
     return when (parsedObjectId.type) {
-      ObjectType.Map -> DefaultLiveMap.zeroValue(objectId, adapter, this) // RTO6b2
-      ObjectType.Counter -> DefaultLiveCounter.zeroValue(objectId, adapter) // RTO6b3
+      ObjectType.Map -> DefaultLiveMap.zeroValue(objectId, liveObjects) // RTO6b2
+      ObjectType.Counter -> DefaultLiveCounter.zeroValue(objectId, liveObjects) // RTO6b3
     }.apply {
       set(objectId, this) // RTO6b4 - Add the zero-value object to the pool
     }

live-objects/src/main/kotlin/io/ably/lib/objects/type/livecounter/DefaultLiveCounter.kt

@@ -6,6 +6,7 @@ import io.ably.lib.objects.ObjectState
 import io.ably.lib.objects.type.BaseLiveObject
 import io.ably.lib.objects.type.ObjectType
 import io.ably.lib.types.Callback
+import java.util.concurrent.atomic.AtomicReference
 
 /**
  * Implementation of LiveObject for LiveCounter.
@@ -14,21 +15,25 @@ import io.ably.lib.types.Callback
  */
 internal class DefaultLiveCounter private constructor(
   objectId: String,
-  adapter: LiveObjectsAdapter,
+  private val liveObjects: DefaultLiveObjects,
 ) : LiveCounter, BaseLiveObject(objectId, ObjectType.Counter) {
 
   override val tag = "LiveCounter"
 
   /**
-   * Counter data value
+   * Thread-safe reference to hold the counter data value.
+   * Accessed from public API for LiveCounter and updated by LiveCounterManager.
    */
-  internal var data: Double = 0.0 // RTLC3
+  internal val data = AtomicReference<Double>(0.0) // RTLC3
 
   /**
    * liveCounterManager instance for managing LiveMap operations
    */
   private val liveCounterManager = LiveCounterManager(this)
 
+  private val channelName = liveObjects.channelName
+  private val adapter: LiveObjectsAdapter get() = liveObjects.adapter
+
   override fun increment() {
     TODO("Not yet implemented")
   }
@@ -60,7 +65,7 @@ internal class DefaultLiveCounter private constructor(
   }
 
   override fun clearData(): Map<String, Double> {
-    return mapOf("amount" to data).apply { data = 0.0 }
+    return mapOf("amount" to data.get()).apply { data.set(0.0) }
   }
 
   override fun onGCInterval() {
@@ -73,8 +78,8 @@ internal class DefaultLiveCounter private constructor(
      * Creates a zero-value counter object.
      * @spec RTLC4 - Returns LiveCounter with 0 value
      */
-    internal fun zeroValue(objectId: String, adapter: LiveObjectsAdapter): DefaultLiveCounter {
-      return DefaultLiveCounter(objectId, adapter)
+    internal fun zeroValue(objectId: String, liveObjects: DefaultLiveObjects): DefaultLiveCounter {
+      return DefaultLiveCounter(objectId, liveObjects)
     }
   }
 }

live-objects/src/main/kotlin/io/ably/lib/objects/type/livecounter/LiveCounterManager.kt

@@ -16,22 +16,22 @@ internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter) {
    * @spec RTLC6 - Overrides counter data with state from sync
    */
   internal fun applyState(objectState: ObjectState): Map<String, Double> {
-    val previousData = liveCounter.data
+    val previousData = liveCounter.data.get()
 
     if (objectState.tombstone) {
       liveCounter.tombstone()
     } else {
       // override data for this object with data from the object state
       liveCounter.createOperationIsMerged = false // RTLC6b
-      liveCounter.data = objectState.counter?.count ?: 0.0 // RTLC6c
+      liveCounter.data.set(objectState.counter?.count ?: 0.0) // RTLC6c
 
       // RTLC6d
       objectState.createOp?.let { createOp ->
         mergeInitialDataFromCreateOperation(createOp)
       }
     }
 
-    return mapOf("amount" to (liveCounter.data - previousData))
+    return mapOf("amount" to (liveCounter.data.get() - previousData))
   }
 
   /**
@@ -78,7 +78,8 @@ internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter) {
    */
   private fun applyCounterInc(counterOp: ObjectCounterOp): Map<String, Double> {
     val amount = counterOp.amount ?: 0.0
-    liveCounter.data += amount // RTLC9b
+    val previousValue = liveCounter.data.get()
+    liveCounter.data.set(previousValue + amount) // RTLC9b
     return mapOf("amount" to amount)
   }
 
@@ -91,7 +92,8 @@ internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter) {
     // if we got here, it means that current counter instance is missing the initial value in its data reference,
     // which we're going to add now.
     val count = operation.counter?.count ?: 0.0
-    liveCounter.data += count // RTLC10a
+    val previousValue = liveCounter.data.get()
+    liveCounter.data.set(previousValue + count) // RTLC10a
     liveCounter.createOperationIsMerged = true // RTLC10b
     return mapOf("amount" to count)
   }

live-objects/src/main/kotlin/io/ably/lib/objects/type/livemap/DefaultLiveMap.kt

@@ -1,34 +1,14 @@
 package io.ably.lib.objects.type.livemap
 
 import io.ably.lib.objects.*
-import io.ably.lib.objects.ObjectsPool
-import io.ably.lib.objects.ObjectsPoolDefaults
 import io.ably.lib.objects.MapSemantics
-import io.ably.lib.objects.ObjectData
 import io.ably.lib.objects.ObjectMessage
 import io.ably.lib.objects.ObjectOperation
 import io.ably.lib.objects.ObjectState
 import io.ably.lib.objects.type.BaseLiveObject
 import io.ably.lib.objects.type.ObjectType
 import io.ably.lib.types.Callback
-
-/**
- * @spec RTLM3 - Map data structure storing entries
- */
-internal data class LiveMapEntry(
-  var isTombstoned: Boolean = false,
-  var tombstonedAt: Long? = null,
-  var timeserial: String? = null,
-  var data: ObjectData? = null
-)
-
-/**
- * Extension function to check if a LiveMapEntry is expired and ready for garbage collection
- */
-private fun LiveMapEntry.isEligibleForGc(): Boolean {
-  val currentTime = System.currentTimeMillis()
-  return isTombstoned && tombstonedAt?.let { currentTime - it >= ObjectsPoolDefaults.GC_GRACE_PERIOD_MS } == true
-}
+import java.util.concurrent.ConcurrentHashMap
 
 /**
  * Implementation of LiveObject for LiveMap.
@@ -37,22 +17,25 @@ private fun LiveMapEntry.isEligibleForGc(): Boolean {
  */
 internal class DefaultLiveMap private constructor(
   objectId: String,
-  adapter: LiveObjectsAdapter,
-  internal val objectsPool: ObjectsPool,
+  private val liveObjects: DefaultLiveObjects,
   internal val semantics: MapSemantics = MapSemantics.LWW
 ) : LiveMap, BaseLiveObject(objectId, ObjectType.Map) {
 
   override val tag = "LiveMap"
+
   /**
-   * Map of key to LiveMapEntry
+   * ConcurrentHashMap for thread-safe access from public APIs in LiveMap and LiveMapManager.
    */
-  internal val data = mutableMapOf<String, LiveMapEntry>()
+  internal val data = ConcurrentHashMap<String, LiveMapEntry>()
 
   /**
    * LiveMapManager instance for managing LiveMap operations
    */
   private val liveMapManager = LiveMapManager(this)
 
+  private val channelName = liveObjects.channelName
+  private val adapter: LiveObjectsAdapter get() = liveObjects.adapter
+  internal val objectsPool: ObjectsPool get() = liveObjects.objectsPool
 
   override fun get(keyName: String): Any? {
     TODO("Not yet implemented")
@@ -114,8 +97,8 @@ internal class DefaultLiveMap private constructor(
      * Creates a zero-value map object.
      * @spec RTLM4 - Returns LiveMap with empty map data
      */
-    internal fun zeroValue(objectId: String, adapter: LiveObjectsAdapter, objectsPool: ObjectsPool): DefaultLiveMap {
-      return DefaultLiveMap(objectId, adapter, objectsPool)
+    internal fun zeroValue(objectId: String, objects: DefaultLiveObjects): DefaultLiveMap {
+      return DefaultLiveMap(objectId, objects)
     }
   }
 }

live-objects/src/main/kotlin/io/ably/lib/objects/type/livemap/LiveMapEntry.kt

@@ -0,0 +1,61 @@
+package io.ably.lib.objects.type.livemap
+
+import io.ably.lib.objects.ObjectData
+import io.ably.lib.objects.ObjectsPool
+import io.ably.lib.objects.ObjectsPoolDefaults
+
+/**
+ * @spec RTLM3 - Map data structure storing entries
+ */
+internal data class LiveMapEntry(
+  val isTombstoned: Boolean = false,
+  val tombstonedAt: Long? = null,
+  val timeserial: String? = null,
+  val data: ObjectData? = null
+)
+
+/**
+ * Checks if entry is directly tombstoned or references a tombstoned object. Spec: RTLM14
+ * @param objectsPool The object pool containing referenced LiveObjects
+ */
+internal fun LiveMapEntry.isEntryOrRefTombstoned(objectsPool: ObjectsPool): Boolean {
+  if (isTombstoned) {
+    return true // RTLM14a
+  }
+  data?.objectId?.let { refId -> // RTLM5d2f -has an objectId reference
+    objectsPool.get(refId)?.let { refObject ->
+      if (refObject.isTombstoned) {
+        return true
+      }
+    }
+  }
+  return false // RTLM14b
+}
+
+/**
+ * Returns value as is if object data stores a primitive type or
+ * a reference to another LiveObject from the pool if it stores an objectId.
+ */
+internal fun LiveMapEntry.getResolvedValue(objectsPool: ObjectsPool): Any? {
+  if (isTombstoned) { return null } // RTLM5d2a
+
+  data?.value?.let { return it.value } // RTLM5d2b, RTLM5d2c, RTLM5d2d, RTLM5d2e
+
+  data?.objectId?.let { refId -> // RTLM5d2f -has an objectId reference
+    objectsPool.get(refId)?.let { refObject ->
+      if (refObject.isTombstoned) {
+        return null // tombstoned objects must not be surfaced to the end users
+      }
+      return refObject // RTLM5d2f2
+    }
+  }
+  return null // RTLM5d2g, RTLM5d2f1
+}
+
+/**
+ * Extension function to check if a LiveMapEntry is expired and ready for garbage collection
+ */
+internal fun LiveMapEntry.isEligibleForGc(): Boolean {
+  val currentTime = System.currentTimeMillis()
+  return isTombstoned && tombstonedAt?.let { currentTime - it >= ObjectsPoolDefaults.GC_GRACE_PERIOD_MS } == true
+}