[ECO-5458] Updated interfaces for mapUpdate and counterUpdate to inherit LiveObjectUpdate

1. Updated mapManager to return LiveMapUpdate instead of Map
2. Updated counterManager to return LiveCounterUpdate instead of map
3. Updated BaseLiveObject accordingly

Author: sacOO7

lib/src/main/java/io/ably/lib/objects/type/LiveObjectUpdate.java

@@ -0,0 +1,9 @@
+package io.ably.lib.objects.type;
+
+public abstract class LiveObjectUpdate {
+    protected final Object update;
+
+    protected LiveObjectUpdate(Object update) {
+        this.update = update;
+    }
+}

lib/src/main/java/io/ably/lib/objects/type/counter/LiveCounter.java

@@ -11,7 +11,7 @@
  * It allows incrementing, decrementing, and retrieving the current value of the counter,
  * both synchronously and asynchronously.
  */
-public interface LiveCounter {
+public interface LiveCounter extends LiveCounterChange {
 
     /**
      * Increments the value of the counter by 1.

lib/src/main/java/io/ably/lib/objects/type/counter/LiveCounterUpdate.java

@@ -1,20 +1,39 @@
 package io.ably.lib.objects.type.counter;
 
+import io.ably.lib.objects.type.LiveObjectUpdate;
 import org.jetbrains.annotations.NotNull;
 
 /**
  * Spec: RTLC11, RTLC11a
  */
-public class LiveCounterUpdate {
-    @NotNull
-    private final Long amount; // RTLC11b, RTLC11b1
+public class LiveCounterUpdate extends LiveObjectUpdate {
+
+    public LiveCounterUpdate() {
+        super(null);
+    }
 
-    public LiveCounterUpdate(@NotNull Long amount) {
-        this.amount = amount;
+    public LiveCounterUpdate(Long amount) {
+        super(new Update(amount));
     }
 
     @NotNull
-    public Long getUpdate() {
-        return amount;
+    public LiveCounterUpdate.Update getUpdate() {
+        return (Update) update;
+    }
+
+    /**
+     * Spec: RTLC11b, RTLC11b1
+     */
+    public static class Update {
+        @NotNull
+        private final Long amount;
+
+        public Update(@NotNull Long amount) {
+            this.amount = amount;
+        }
+
+        public @NotNull Long getAmount() {
+            return amount;
+        }
     }
 }

lib/src/main/java/io/ably/lib/objects/type/map/LiveMap.java

@@ -14,7 +14,7 @@
  * The LiveMap interface provides methods to interact with a live, real-time map structure.
  * It supports both synchronous and asynchronous operations for managing key-value pairs.
  */
-public interface LiveMap {
+public interface LiveMap extends LiveMapChange {
 
     /**
      * Retrieves the value associated with the specified key.

lib/src/main/java/io/ably/lib/objects/type/map/LiveMapUpdate.java

@@ -1,23 +1,25 @@
 package io.ably.lib.objects.type.map;
 
+import io.ably.lib.objects.type.LiveObjectUpdate;
 import org.jetbrains.annotations.NotNull;
 
 import java.util.Map;
 
 /**
  * Spec: RTLM18, RTLM18a
  */
-public class LiveMapUpdate {
+public class LiveMapUpdate extends LiveObjectUpdate {
 
-    @NotNull
-    private final Map<String, Change> update;
+    public LiveMapUpdate() {
+        super(null);
+    }
 
     /**
      * Constructor for LiveMapUpdate
      * @param update The map of updates
      */
     public LiveMapUpdate(@NotNull Map<String, Change> update) {
-        this.update = update;
+        super(update);
     }
 
     /**
@@ -26,7 +28,7 @@ public LiveMapUpdate(@NotNull Map<String, Change> update) {
      */
     @NotNull
     public Map<String, Change> getUpdate() {
-        return update;
+        return (Map<String, Change>) update;
     }
 
     /**

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

@@ -1,6 +1,7 @@
 package io.ably.lib.objects
 
 import io.ably.lib.objects.type.BaseLiveObject
+import io.ably.lib.objects.type.LiveObjectUpdate
 import io.ably.lib.objects.type.livecounter.DefaultLiveCounter
 import io.ably.lib.objects.type.livemap.DefaultLiveMap
 import io.ably.lib.util.Log
@@ -126,7 +127,7 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects): Obje
 
     val receivedObjectIds = mutableSetOf<String>()
     // RTO5c1a2 - List to collect updates for existing objects
-    val existingObjectUpdates = mutableListOf<Pair<BaseLiveObject, Any>>()
+    val existingObjectUpdates = mutableListOf<Pair<BaseLiveObject, LiveObjectUpdate>>()
 
     // RTO5c1
     for ((objectId, objectState) in syncObjectsDataPool) {

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

@@ -6,13 +6,17 @@ 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")
 }
 
+internal val LiveObjectUpdate.noOp get() = this.update == null
+
 /**
  * Base implementation of LiveObject interface.
  * Provides common functionality for all live objects.
@@ -43,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 {
     if (objectState.objectId != objectId) {
       throw objectError("Invalid object state: object state objectId=${objectState.objectId}; $objectType objectId=$objectId")
     }
@@ -61,7 +65,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
   }
@@ -102,11 +109,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.
    *
@@ -126,7 +128,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()
@@ -151,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.
@@ -177,7 +179,13 @@ 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.
+   */
+  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,9 +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 io.ably.lib.types.Callback
+import io.ably.lib.util.Log
 import java.util.concurrent.atomic.AtomicLong
 
 /**
@@ -50,21 +55,38 @@ internal class DefaultLiveCounter private constructor(
     TODO("Not yet implemented")
   }
 
+  override fun subscribe(listener: LiveCounterChange.Listener): ObjectsSubscription {
+    return liveCounterManager.subscribe(listener)
+  }
+
+  override fun unsubscribe(listener: LiveCounterChange.Listener) = liveCounterManager.unsubscribe(listener)
+
+  override fun unsubscribeAll() = liveCounterManager.unsubscribeAll()
+
   override fun value(): Long {
     adapter.throwIfInvalidAccessApiConfiguration(channelName)
     return data.get()
   }
 
-  override fun applyObjectState(objectState: ObjectState): Map<String, Long> {
+  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, Long> {
-    return mapOf("amount" to data.get()).apply { data.set(0) }
+  override fun clearData(): LiveCounterUpdate {
+    return LiveCounterUpdate(data.get()).apply { data.set(0) }
+  }
+
+  override fun notifyUpdated(update: LiveObjectUpdate) {
+    if (update.noOp) {
+      return
+    }
+    val counterUpdate = update as LiveCounterUpdate
+    Log.v(tag, "Object $objectId updated: $update")
+    liveCounterManager.notify(counterUpdate)
   }
 
   override fun onGCInterval() {

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

@@ -0,0 +1,44 @@
+package io.ably.lib.objects.type.livecounter
+
+import io.ably.lib.objects.ObjectsSubscription
+import io.ably.lib.objects.type.counter.LiveCounterChange
+import io.ably.lib.objects.type.counter.LiveCounterUpdate
+import io.ably.lib.util.EventEmitter
+import io.ably.lib.util.Log
+
+internal val noOpCounterUpdate = LiveCounterUpdate()
+
+internal interface HandlesLiveCounterChange {
+  fun notify(update: LiveCounterUpdate)
+}
+
+internal abstract class LiveCounterChangeCoordinator: LiveCounterChange, HandlesLiveCounterChange {
+  private val tag = "DefaultLiveCounterChangeCoordinator"
+
+  private val counterChangeEmitter = LiveCounterChangeEmitter()
+
+  override fun subscribe(listener: LiveCounterChange.Listener): ObjectsSubscription {
+    counterChangeEmitter.on(listener)
+    return ObjectsSubscription {
+      counterChangeEmitter.off(listener)
+    }
+  }
+
+  override fun unsubscribe(listener: LiveCounterChange.Listener) = counterChangeEmitter.off(listener)
+
+  override fun unsubscribeAll() = counterChangeEmitter.off()
+
+  override fun notify(update: LiveCounterUpdate) = counterChangeEmitter.emit(update)
+}
+
+private class LiveCounterChangeEmitter : EventEmitter<LiveCounterUpdate, LiveCounterChange.Listener>() {
+  private val tag = "LiveCounterChangeEmitter"
+
+  override fun apply(listener: LiveCounterChange.Listener?, event: LiveCounterUpdate?, vararg args: Any?) {
+    try {
+      listener?.onUpdated(event!!)
+    } catch (t: Throwable) {
+      Log.e(tag, "Error occurred while executing listener callback for event: $event", t)
+    }
+  }
+}

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, Long> {
+  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)
   }
 
   /**
@@ -57,7 +59,7 @@ internal class LiveCounterManager(private val liveCounter: DefaultLiveCounter) {
   /**
    * @spec RTLC8 - Applies counter create operation
    */
-  private fun applyCounterCreate(operation: ObjectOperation): Map<String, Long> {
+  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, Long> {
+  private fun applyCounterInc(counterOp: ObjectCounterOp): LiveCounterUpdate {
     val amount = counterOp.amount?.toLong() ?: 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, Long> {
+  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,6 +97,6 @@ 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)
   }
 }