[ECO-5457] feat: Add Live Objects state change handling

• Add ObjectsStateEvent enum for tracking Live Objects synchronization states (SYNCING/SYNCED)
• Implement ObjectsStateListener interface for receiving state change notifications
• Create ObjectsStateSubscription interface for managing listener subscriptions and cleanup
• Update LiveObjects, DefaultLiveObjects, ObjectsManager, ObjectsState, and Utils classes to support state management

Author: sacOO7

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

@@ -1,5 +1,8 @@
 package io.ably.lib.objects;
 
+import io.ably.lib.objects.state.ObjectsStateEvent;
+import io.ably.lib.objects.state.ObjectsStateListener;
+import io.ably.lib.objects.state.ObjectsStateSubscription;
 import io.ably.lib.types.Callback;
 import org.jetbrains.annotations.Blocking;
 import org.jetbrains.annotations.NonBlocking;
@@ -148,4 +151,38 @@ public interface LiveObjects {
      */
     @NonBlocking
     void createCounterAsync(@NotNull Long initialValue, @NotNull Callback<@NotNull LiveCounter> callback);
+
+    /**
+     * Subscribes to a specific Live Objects synchronization state event.
+     *
+     * <p>This method registers the provided listener to be notified when the specified
+     * synchronization state event occurs. The returned subscription can be used to
+     * unsubscribe later when the notifications are no longer needed.
+     *
+     * @param event the synchronization state event to subscribe to (SYNCING or SYNCED)
+     * @param listener the listener that will be called when the event occurs
+     * @return a subscription object that can be used to unsubscribe from the event
+     */
+    @NonBlocking
+    ObjectsStateSubscription on(@NotNull ObjectsStateEvent event, @NotNull ObjectsStateListener listener);
+
+    /**
+     * Unsubscribes the specified listener from all synchronization state events.
+     *
+     * <p>After calling this method, the provided listener will no longer receive
+     * any synchronization state event notifications.
+     *
+     * @param listener the listener to unregister from all events
+     */
+    @NonBlocking
+    void off(@NotNull ObjectsStateListener listener);
+
+    /**
+     * Unsubscribes all listeners from all synchronization state events.
+     *
+     * <p>After calling this method, no listeners will receive any synchronization
+     * state event notifications until new listeners are registered.
+     */
+    @NonBlocking
+    void offAll();
 }

lib/src/main/java/io/ably/lib/objects/state/ObjectsStateEvent.java

@@ -0,0 +1,19 @@
+package io.ably.lib.objects.state;
+
+/**
+ * Represents the synchronization state of Ably Live Objects.
+ * <p>
+ * This enum is used to notify listeners about state changes in the synchronization process.
+ * Clients can register an {@link ObjectsStateListener} to receive these events.
+ */
+public enum ObjectsStateEvent {
+    /**
+     * Indicates that synchronization between local and remote objects is in progress.
+     */
+    SYNCING,
+
+    /**
+     * Indicates that synchronization has completed successfully and objects are in sync.
+     */
+    SYNCED
+}

lib/src/main/java/io/ably/lib/objects/state/ObjectsStateListener.java

@@ -0,0 +1,16 @@
+package io.ably.lib.objects.state;
+
+/**
+ * Interface for receiving notifications about Live Objects synchronization state changes.
+ * <p>
+ * Implement this interface and register it with an ObjectsStateEmitter to be notified
+ * when synchronization state transitions occur.
+ */
+public interface ObjectsStateListener {
+    /**
+     * Called when the synchronization state changes.
+     *
+     * @param objectsStateEvent The new state event (SYNCING or SYNCED)
+     */
+    void onStateChanged(ObjectsStateEvent objectsStateEvent);
+}

lib/src/main/java/io/ably/lib/objects/state/ObjectsStateSubscription.java

@@ -0,0 +1,20 @@
+package io.ably.lib.objects.state;
+
+/**
+ * Represents a subscription that can be unsubscribed from.
+ * This interface provides a way to clean up and remove subscriptions when they
+ * are no longer needed.
+ * Example usage:
+ * ```java
+ * ObjectsStateSubscription s = objects.subscribe(ObjectsStateEvent.SYNCING, new ObjectsStateListener() {});
+ * // Later when done with the subscription
+ * s.unsubscribe();
+ */
+public interface ObjectsStateSubscription {
+    /**
+     * This method should be called when the subscription is no longer needed,
+     * it will make sure no further events will be sent to the subscriber and
+     * that references to the subscriber are cleaned up.
+     */
+    void unsubscribe();
+}

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

@@ -1,5 +1,8 @@
 package io.ably.lib.objects
 
+import io.ably.lib.objects.state.ObjectsStateEvent
+import io.ably.lib.objects.state.ObjectsStateListener
+import io.ably.lib.objects.state.ObjectsStateSubscription
 import io.ably.lib.realtime.ChannelState
 import io.ably.lib.types.Callback
 import io.ably.lib.types.ProtocolMessage
@@ -52,7 +55,7 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
   }
 
   override fun getRootAsync(callback: Callback<LiveMap>) {
-    callbackScope.with(callback) { getRootAsync() }
+    callbackScope.launchWithCallback(callback) { getRootAsync() }
   }
 
   override fun createMap(liveMap: LiveMap): LiveMap {
@@ -87,6 +90,21 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
     TODO("Not yet implemented")
   }
 
+  override fun on(event: ObjectsStateEvent, listener: ObjectsStateListener): ObjectsStateSubscription {
+    objectsManager.publicObjectStateEmitter.on(event, listener)
+    return ObjectsStateSubscription {
+      objectsManager.publicObjectStateEmitter.off(event, listener)
+    }
+  }
+
+  override fun off(listener: ObjectsStateListener) {
+    objectsManager.publicObjectStateEmitter.off(listener)
+  }
+
+  override fun offAll() {
+    objectsManager.publicObjectStateEmitter.off()
+  }
+
   private suspend fun getRootAsync(): LiveMap {
     return sequentialScope.async {
       adapter.throwIfInvalidAccessApiConfiguration(channelName)

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

@@ -1,5 +1,6 @@
 package io.ably.lib.objects
 
+import io.ably.lib.objects.state.ObjectsStateEvent
 import io.ably.lib.objects.type.BaseLiveObject
 import io.ably.lib.objects.type.livecounter.DefaultLiveCounter
 import io.ably.lib.objects.type.livemap.DefaultLiveMap
@@ -25,7 +26,7 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
   // composition over inheritance, used to handle object state changes internally
   private val internalObjectStateEmitter = ObjectsStateEmitter()
   // related to RTC10, should have a separate EventEmitter for users of the library
-  private val publicObjectStateEmitter = ObjectsStateEmitter()
+  internal val publicObjectStateEmitter = ObjectsStateEmitter()
   // Coroutine scope for running sequential operations on a single thread, used to avoid concurrency issues.
   private val emitterScope = CoroutineScope(Dispatchers.Default.limitedParallelism(1) + SupervisorJob())
 
@@ -230,7 +231,7 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
   internal suspend fun ensureSynced() {
     if (liveObjects.state != ObjectsState.SYNCED) {
       val deferred = CompletableDeferred<Unit>()
-      internalObjectStateEmitter.once(ObjectsEvent.SYNCED) {
+      internalObjectStateEmitter.once(ObjectsStateEvent.SYNCED) {
         Log.v(tag, "Objects state changed to SYNCED, resuming ensureSynced")
         deferred.complete(Unit)
       }
@@ -253,8 +254,8 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
 
     val event = objectsStateToEventMap[newState]
     event?.let {
-      // Use of deferEvent not needed, since emit method is synchronized amongst different threads
-      // emitterScope makes sure, next launch can only start when previous launch finishes processing of all events
+      // Use of deferEvent not needed since emitterScope makes sure next launch can only start when previous launch
+      // finishes processing of all events. Also, emit method is synchronized amongst different threads
       emitterScope.launch {
         internalObjectStateEmitter.emit(it)
         publicObjectStateEmitter.emit(it)

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

@@ -1,5 +1,7 @@
 package io.ably.lib.objects
 
+import io.ably.lib.objects.state.ObjectsStateEvent
+import io.ably.lib.objects.state.ObjectsStateListener
 import io.ably.lib.util.EventEmitter
 
 /**
@@ -11,23 +13,14 @@ internal enum class ObjectsState {
   SYNCED
 }
 
-public enum class ObjectsEvent {
-  SYNCING,
-  SYNCED
-}
-
 internal val objectsStateToEventMap = mapOf(
   ObjectsState.INITIALIZED to null,
-  ObjectsState.SYNCING to ObjectsEvent.SYNCING,
-  ObjectsState.SYNCED to ObjectsEvent.SYNCED
+  ObjectsState.SYNCING to ObjectsStateEvent.SYNCING,
+  ObjectsState.SYNCED to ObjectsStateEvent.SYNCED
 )
 
-public fun interface ObjectsStateListener {
-  public fun onStateChanged(state: ObjectsEvent)
-}
-
-internal class ObjectsStateEmitter : EventEmitter<ObjectsEvent, ObjectsStateListener>() {
-  override fun apply(listener: ObjectsStateListener?, event: ObjectsEvent?, vararg args: Any?) {
+internal class ObjectsStateEmitter : EventEmitter<ObjectsStateEvent, ObjectsStateListener>() {
+  override fun apply(listener: ObjectsStateListener?, event: ObjectsStateEvent?, vararg args: Any?) {
     listener?.onStateChanged(event!!)
   }
 }

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

@@ -58,7 +58,7 @@ internal val String.byteSize: Int
  * @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
  */
-internal fun <T> CoroutineScope.with(callback: Callback<T>, block: suspend () -> T) {
+internal fun <T> CoroutineScope.launchWithCallback(callback: Callback<T>, block: suspend () -> T) {
   launch {
     try {
       val result = block()