[ECO-5458] Enhanced LiveCounter with subscription support

- Updated BaseLiveObject with subscription coordination capabilities
- Modified DefaultLiveCounter to integrate with LiveCounterChangeCoordinator
- Enhanced LiveCounterManager to return LiveCounterUpdate objects

Author: sacOO7

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

@@ -5,13 +5,18 @@ import io.ably.lib.objects.ObjectOperation
 import io.ably.lib.objects.ObjectState
 import io.ably.lib.objects.ObjectsPoolDefaults
 import io.ably.lib.objects.objectError
+import io.ably.lib.objects.type.livecounter.noOpCounterUpdate
+import io.ably.lib.objects.type.livemap.noOpMapUpdate
 import io.ably.lib.util.Log
 
 internal enum class ObjectType(val value: String) {
   Map("map"),
   Counter("counter")
 }
 
+// Spec: RTLO4b4b
+internal val LiveObjectUpdate.noOp get() = this.update == null
+
 /**
  * Base implementation of LiveObject interface.
  * Provides common functionality for all live objects.
@@ -42,7 +47,7 @@ internal abstract class BaseLiveObject(
    *
    * @spec RTLM6/RTLC6 - Overrides ObjectMessage with object data state from sync to LiveMap/LiveCounter
    */
-  internal fun applyObjectSync(objectState: ObjectState): Map<String, Any> {
+  internal fun applyObjectSync(objectState: ObjectState): LiveObjectUpdate {
     validate(objectState)
     // object's site serials are still updated even if it is tombstoned, so always use the site serials received from the operation.
     // should default to empty map if site serials do not exist on the object state, so that any future operation may be applied to this object.
@@ -51,7 +56,10 @@ internal abstract class BaseLiveObject(
 
     if (isTombstoned) {
       // this object is tombstoned. this is a terminal state which can't be overridden. skip the rest of object state message processing
-      return mapOf()
+      if (objectType == ObjectType.Map) {
+        return noOpMapUpdate
+      }
+      return noOpCounterUpdate
     }
     return applyObjectState(objectState) // RTLM6, RTLC6
   }
@@ -74,7 +82,7 @@ internal abstract class BaseLiveObject(
       Log.v(
         tag,
         "Skipping ${objectOperation.action} op: op serial $msgTimeSerial <= site serial ${siteTimeserials[msgSiteCode]}; " +
-                "objectId=$objectId"
+          "objectId=$objectId"
       )
       return
     }
@@ -89,11 +97,6 @@ internal abstract class BaseLiveObject(
     applyObjectOperation(objectOperation, objectMessage) // RTLC7d
   }
 
-  internal fun notifyUpdated(update: Any) {
-    // TODO: Implement event emission for updates
-    Log.v(tag, "Object $objectId updated: $update")
-  }
-
   /**
    * Checks if an operation can be applied based on serial comparison.
    *
@@ -119,7 +122,7 @@ internal abstract class BaseLiveObject(
   /**
    * Marks the object as tombstoned.
    */
-  internal fun tombstone(): Any {
+  internal fun tombstone(): LiveObjectUpdate {
     isTombstoned = true
     tombstonedAt = System.currentTimeMillis()
     val update = clearData()
@@ -150,7 +153,7 @@ internal abstract class BaseLiveObject(
    * @return A map describing the changes made to the object's data
    *
    */
-  abstract fun applyObjectState(objectState: ObjectState): Map<String, Any>
+  abstract fun applyObjectState(objectState: ObjectState): LiveObjectUpdate
 
   /**
    * Applies an operation to this live object.
@@ -176,7 +179,14 @@ internal abstract class BaseLiveObject(
    *
    * @return A map representing the diff of changes made
    */
-  abstract fun clearData(): Map<String, Any>
+  abstract fun clearData(): LiveObjectUpdate
+
+  /**
+   * Notifies subscribers about changes made to this live object. Propagates updates through the
+   * appropriate manager after converting the generic update map to type-specific update objects.
+   * Spec: RTLO4b4c
+   */
+  abstract fun notifyUpdated(update: LiveObjectUpdate)
 
   /**
    * Called during garbage collection intervals to clean up expired entries.

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

@@ -4,8 +4,14 @@ import io.ably.lib.objects.*
 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.LiveObjectUpdate
 import io.ably.lib.objects.type.ObjectType
+import io.ably.lib.objects.type.counter.LiveCounter
+import io.ably.lib.objects.type.counter.LiveCounterChange
+import io.ably.lib.objects.type.counter.LiveCounterUpdate
+import io.ably.lib.objects.type.noOp
 import java.util.concurrent.atomic.AtomicReference
+import io.ably.lib.util.Log
 
 /**
  * Implementation of LiveObject for LiveCounter.
@@ -54,18 +60,35 @@ internal class DefaultLiveCounter private constructor(
     return data.get()
   }
 
+  override fun subscribe(listener: LiveCounterChange.Listener): ObjectsSubscription {
+    adapter.throwIfInvalidAccessApiConfiguration(channelName)
+    return liveCounterManager.subscribe(listener)
+  }
+
+  override fun unsubscribe(listener: LiveCounterChange.Listener) = liveCounterManager.unsubscribe(listener)
+
+  override fun unsubscribeAll() = liveCounterManager.unsubscribeAll()
+
   override fun validate(state: ObjectState) = liveCounterManager.validate(state)
 
-  override fun applyObjectState(objectState: ObjectState): Map<String, Double> {
+  override fun applyObjectState(objectState: ObjectState): LiveCounterUpdate {
     return liveCounterManager.applyState(objectState)
   }
 
   override fun applyObjectOperation(operation: ObjectOperation, message: ObjectMessage) {
     liveCounterManager.applyOperation(operation)
   }
 
-  override fun clearData(): Map<String, Double> {
-    return mapOf("amount" to data.get()).apply { data.set(0.0) }
+  override fun clearData(): LiveCounterUpdate {
+    return LiveCounterUpdate(data.get()).apply { data.set(0.0) }
+  }
+
+  override fun notifyUpdated(update: LiveObjectUpdate) {
+    if (update.noOp) {
+      return
+    }
+    Log.v(tag, "Object $objectId updated: $update")
+    liveCounterManager.notify(update as LiveCounterUpdate)
   }
 
   override fun onGCInterval() {

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

@@ -5,17 +5,19 @@ import io.ably.lib.objects.ObjectOperation
 import io.ably.lib.objects.ObjectOperationAction
 import io.ably.lib.objects.ObjectState
 import io.ably.lib.objects.objectError
+import io.ably.lib.objects.type.counter.LiveCounterUpdate
 import io.ably.lib.util.Log
 
-internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter) {
+internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter): LiveCounterChangeCoordinator() {
+
   private val objectId = liveCounter.objectId
 
   private val tag = "LiveCounterManager"
 
   /**
    * @spec RTLC6 - Overrides counter data with state from sync
    */
-  internal fun applyState(objectState: ObjectState): Map<String, Double> {
+  internal fun applyState(objectState: ObjectState): LiveCounterUpdate {
     val previousData = liveCounter.data.get()
 
     if (objectState.tombstone) {
@@ -31,7 +33,7 @@ internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter) {
       }
     }
 
-    return mapOf("amount" to (liveCounter.data.get() - previousData))
+    return LiveCounterUpdate(liveCounter.data.get() - previousData)
   }
 
   /**
@@ -51,13 +53,13 @@ internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter) {
       else -> throw objectError("Invalid ${operation.action} op for LiveCounter objectId=${objectId}") // RTLC7d3
     }
 
-    liveCounter.notifyUpdated(update)
+    liveCounter.notifyUpdated(update) // RTLC7d1a, RTLC7d2a
   }
 
   /**
    * @spec RTLC8 - Applies counter create operation
    */
-  private fun applyCounterCreate(operation: ObjectOperation): Map<String, Double> {
+  private fun applyCounterCreate(operation: ObjectOperation): LiveCounterUpdate {
     if (liveCounter.createOperationIsMerged) {
       // RTLC8b
       // There can't be two different create operation for the same object id, because the object id
@@ -67,7 +69,7 @@ internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter) {
         tag,
         "Skipping applying COUNTER_CREATE op on a counter instance as it was already applied before; objectId=$objectId"
       )
-      return mapOf()
+      return noOpCounterUpdate // RTLC8c
     }
 
     return mergeInitialDataFromCreateOperation(operation) // RTLC8c
@@ -76,17 +78,17 @@ internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter) {
   /**
    * @spec RTLC9 - Applies counter increment operation
    */
-  private fun applyCounterInc(counterOp: ObjectCounterOp): Map<String, Double> {
+  private fun applyCounterInc(counterOp: ObjectCounterOp): LiveCounterUpdate {
     val amount = counterOp.amount ?: 0.0
     val previousValue = liveCounter.data.get()
     liveCounter.data.set(previousValue + amount) // RTLC9b
-    return mapOf("amount" to amount)
+    return LiveCounterUpdate(amount)
   }
 
   /**
    * @spec RTLC10 - Merges initial data from create operation
    */
-  private fun mergeInitialDataFromCreateOperation(operation: ObjectOperation): Map<String, Double> {
+  private fun mergeInitialDataFromCreateOperation(operation: ObjectOperation): LiveCounterUpdate {
     // if a counter object is missing for the COUNTER_CREATE op, the initial value is implicitly 0 in this case.
     // note that it is intentional to SUM the incoming count from the create op.
     // if we got here, it means that current counter instance is missing the initial value in its data reference,
@@ -95,7 +97,7 @@ internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter) {
     val previousValue = liveCounter.data.get()
     liveCounter.data.set(previousValue + count) // RTLC10a
     liveCounter.createOperationIsMerged = true // RTLC10b
-    return mapOf("amount" to count)
+    return LiveCounterUpdate(count)
   }
 
   internal fun validate(state: ObjectState) {