[ECO-5457] Created separate interface for ObjectsStateChange to avoid polluting

LiveObjects
- Added impl. for the same in ObjectsState.kt
- Extended the impl. in objectsmanager

Author: sacOO7

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

@@ -1,8 +1,6 @@
 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.objects.state.ObjectsStateChange;
 import io.ably.lib.types.Callback;
 import org.jetbrains.annotations.Blocking;
 import org.jetbrains.annotations.NonBlocking;
@@ -19,7 +17,7 @@
  * <p>Implementations of this interface must be thread-safe as they may be accessed
  * from multiple threads concurrently.
  */
-public interface LiveObjects {
+public interface LiveObjects extends ObjectsStateChange {
 
     /**
      * Retrieves the root LiveMap object.
@@ -151,38 +149,4 @@ 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/ObjectsStateChange.java

@@ -0,0 +1,55 @@
+package io.ably.lib.objects.state;
+
+import org.jetbrains.annotations.NonBlocking;
+import org.jetbrains.annotations.NotNull;
+
+public interface ObjectsStateChange {
+    /**
+     * 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 ObjectsStateChange.Listener 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 ObjectsStateChange.Listener 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();
+
+    /**
+     * 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.
+     */
+    interface Listener {
+        /**
+         * 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/ObjectsStateEvent.java

@@ -4,7 +4,7 @@
  * 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.
+ * Clients can register an {@link ObjectsStateChange.Listener} to receive these events.
  */
 public enum ObjectsStateEvent {
     /**

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

@@ -1,7 +1,7 @@
 package io.ably.lib.objects
 
+import io.ably.lib.objects.state.ObjectsStateChange
 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
@@ -90,25 +90,17 @@ 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 on(event: ObjectsStateEvent, listener: ObjectsStateChange.Listener): ObjectsStateSubscription =
+    objectsManager.on(event, listener)
 
-  override fun off(listener: ObjectsStateListener) {
-    objectsManager.publicObjectStateEmitter.off(listener)
-  }
+  override fun off(listener: ObjectsStateChange.Listener) = objectsManager.off(listener)
 
-  override fun offAll() {
-    objectsManager.publicObjectStateEmitter.off()
-  }
+  override fun offAll() = objectsManager.offAll()
 
   private suspend fun getRootAsync(): LiveMap {
     return sequentialScope.async {
       adapter.throwIfInvalidAccessApiConfiguration(channelName)
-      objectsManager.ensureSynced()
+      objectsManager.ensureSynced(state)
       objectsPool.get(ROOT_OBJECT_ID) as LiveMap
     }.await()
   }

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

@@ -1,17 +1,15 @@
 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
 import io.ably.lib.util.Log
-import kotlinx.coroutines.*
 
 /**
  * @spec RTO5 - Processes OBJECT and OBJECT_SYNC messages during sync sequences
  * @spec RTO6 - Creates zero-value objects when needed
  */
-internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
+internal class ObjectsManager(private val liveObjects: DefaultLiveObjects): ObjectsStateCoordinator() {
   private val tag = "ObjectsManager"
   /**
    * @spec RTO5 - Sync objects data pool for collecting sync messages
@@ -23,13 +21,6 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
    */
   private val bufferedObjectOperations = mutableListOf<ObjectMessage>() // RTO7a
 
-  // 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
-  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())
-
   /**
    * Handles object messages (non-sync messages).
    *
@@ -224,21 +215,6 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
     }
   }
 
-  /**
-   * Suspends the current coroutine until objects are synchronized.
-   * Returns immediately if state is already SYNCED, otherwise waits for the SYNCED event.
-   */
-  internal suspend fun ensureSynced() {
-    if (liveObjects.state != ObjectsState.SYNCED) {
-      val deferred = CompletableDeferred<Unit>()
-      internalObjectStateEmitter.once(ObjectsStateEvent.SYNCED) {
-        Log.v(tag, "Objects state changed to SYNCED, resuming ensureSynced")
-        deferred.complete(Unit)
-      }
-      deferred.await()
-    }
-  }
-
   /**
    * Changes the state and emits events.
    *
@@ -248,24 +224,16 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
     if (liveObjects.state == newState) {
       return
     }
-
+    Log.v(tag, "Objects state changed to: $newState from ${liveObjects.state}")
     liveObjects.state = newState
-    Log.v(tag, "Objects state changed to: $newState")
 
-    val event = objectsStateToEventMap[newState]
-    event?.let {
-      // 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)
-      }
-    }
+    // deferEvent not needed since objectsStateChanged processes events in a sequential coroutine scope
+    objectsStateChanged(newState)
   }
 
   internal fun dispose() {
     syncObjectsDataPool.clear()
     bufferedObjectOperations.clear()
-    emitterScope.cancel("ObjectsManager disposed")
+    disposeObjectsStateListeners()
   }
 }

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

@@ -1,8 +1,11 @@
 package io.ably.lib.objects
 
+import io.ably.lib.objects.state.ObjectsStateChange
 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.util.EventEmitter
+import io.ably.lib.util.Log
+import kotlinx.coroutines.*
 
 /**
  * @spec RTO2 - enum representing objects state
@@ -13,14 +16,96 @@ internal enum class ObjectsState {
   SYNCED
 }
 
-internal val objectsStateToEventMap = mapOf(
+/**
+ * Maps internal ObjectsState values to their corresponding public ObjectsStateEvent values.
+ * Used to determine which events should be emitted when state changes occur.
+ * 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
 )
 
-internal class ObjectsStateEmitter : EventEmitter<ObjectsStateEvent, ObjectsStateListener>() {
-  override fun apply(listener: ObjectsStateListener?, event: ObjectsStateEvent?, vararg args: Any?) {
+/**
+ * An interface for managing and communicating changes in the synchronization state of live objects.
+ *
+ * Implementations should ensure thread-safe event emission and proper synchronization
+ * between state change notifications.
+ */
+internal interface HandlesObjectsStateChange {
+  /**
+   * Handles changes in the state of live objects by notifying all registered listeners.
+   * Implementations should ensure thread-safe event emission to both internal and public listeners.
+   * Makes sure every event is processed in the order they were received.
+   * @param newState The new state of the objects, SYNCING or SYNCED.
+   */
+  fun objectsStateChanged(newState: ObjectsState)
+
+  /**
+   * Suspends the current coroutine until objects are synchronized.
+   * Returns immediately if state is already SYNCED, otherwise waits for the SYNCED event.
+   *
+   * @param currentState The current state of objects to determine if waiting is necessary
+   */
+  suspend fun ensureSynced(currentState: ObjectsState)
+
+  /**
+   * Disposes all registered state change listeners and cancels any pending operations.
+   * Should be called when the associated LiveObjects instance is no longer needed.
+   */
+  fun disposeObjectsStateListeners()
+}
+
+
+internal abstract class ObjectsStateCoordinator : ObjectsStateChange, HandlesObjectsStateChange {
+  private val tag = "ObjectsStateCoordinator"
+  private val internalObjectStateEmitter = ObjectsStateEmitter()
+  // 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): ObjectsStateSubscription {
+    externalObjectStateEmitter.on(event, listener)
+    return ObjectsStateSubscription {
+      externalObjectStateEmitter.off(event, listener)
+    }
+  }
+
+  override fun off(listener: ObjectsStateChange.Listener) = externalObjectStateEmitter.off(listener)
+
+  override fun offAll() = externalObjectStateEmitter.off()
+
+  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)
+      }
+    }
+  }
+
+  override suspend fun ensureSynced(currentState: ObjectsState) {
+    if (currentState != ObjectsState.SYNCED) {
+      val deferred = CompletableDeferred<Unit>()
+      internalObjectStateEmitter.once(ObjectsStateEvent.SYNCED) {
+        Log.v(tag, "Objects state changed to SYNCED, resuming ensureSynced")
+        deferred.complete(Unit)
+      }
+      deferred.await()
+    }
+  }
+
+  override fun disposeObjectsStateListeners() {
+    offAll()
+    emitterScope.cancel("ObjectsManager disposed")
+  }
+}
+
+private class ObjectsStateEmitter : EventEmitter<ObjectsStateEvent, ObjectsStateChange.Listener>() {
+  override fun apply(listener: ObjectsStateChange.Listener?, event: ObjectsStateEvent?, vararg args: Any?) {
     listener?.onStateChanged(event!!)
   }
 }