[ECO-5076] Moved ObjectsState to separate file

1. Added ObjectsEvents and emitter for the same
2. Implemented stateChange mechanism for objectsState changes
3. updated getRootAsync to ensure objects are synced

Author: sacOO7

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

@@ -8,15 +8,6 @@ import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import kotlinx.coroutines.flow.MutableSharedFlow
 
-/**
- * @spec RTO2 - enum representing objects state
- */
-internal enum class ObjectsState {
-  INITIALIZED,
-  SYNCING,
-  SYNCED
-}
-
 /**
  * Default implementation of LiveObjects interface.
  * Provides the core functionality for managing live objects on a channel.
@@ -99,7 +90,7 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
   private suspend fun getRootAsync(): LiveMap {
     return sequentialScope.async {
       adapter.throwIfInvalidAccessApiConfiguration(channelName)
-      // TODO - wait for state in synced state
+      objectsManager.ensureSynced()
       objectsPool.get(ROOT_OBJECT_ID) as LiveMap
     }.await()
   }
@@ -186,22 +177,6 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
     }
   }
 
-  /**
-   * Changes the state and emits events.
-   *
-   * @spec RTO2 - Emits state change events for syncing and synced states
-   */
-  internal fun stateChange(newState: ObjectsState, deferEvent: Boolean) {
-    if (state == newState) {
-      return
-    }
-
-    state = newState
-    Log.v(tag, "Objects state changed to: $newState")
-
-    // TODO: Emit state change events
-  }
-
   // Dispose of any resources associated with this LiveObjects instance
   fun dispose() {
     incomingObjectsHandler.cancel() // objectsEventBus automatically garbage collected when collector is cancelled

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

@@ -4,6 +4,7 @@ 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
@@ -21,6 +22,13 @@ 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
+  private 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).
    *
@@ -77,7 +85,7 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
     bufferedObjectOperations.clear() // RTO5a2b
     syncObjectsDataPool.clear() // RTO5a2a
     currentSyncId = syncId
-    liveObjects.stateChange(ObjectsState.SYNCING, false)
+    stateChange(ObjectsState.SYNCING, false)
   }
 
   /**
@@ -95,7 +103,7 @@ internal class ObjectsManager(private val liveObjects: DefaultLiveObjects) {
     bufferedObjectOperations.clear() // RTO5c5
     syncObjectsDataPool.clear() // RTO5c4
     currentSyncId = null // RTO5c3
-    liveObjects.stateChange(ObjectsState.SYNCED, deferStateEvent)
+    stateChange(ObjectsState.SYNCED, deferStateEvent)
   }
 
   /**
@@ -215,8 +223,48 @@ 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(ObjectsEvent.SYNCED) {
+        Log.v(tag, "Objects state changed to SYNCED, resuming ensureSynced")
+        deferred.complete(Unit)
+      }
+      deferred.await()
+    }
+  }
+
+  /**
+   * Changes the state and emits events.
+   *
+   * @spec RTO2 - Emits state change events for syncing and synced states
+   */
+  private fun stateChange(newState: ObjectsState, deferEvent: Boolean) {
+    if (liveObjects.state == newState) {
+      return
+    }
+
+    liveObjects.state = newState
+    Log.v(tag, "Objects state changed to: $newState")
+
+    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
+      emitterScope.launch {
+        internalObjectStateEmitter.emit(it)
+        publicObjectStateEmitter.emit(it)
+      }
+    }
+  }
+
   internal fun dispose() {
     syncObjectsDataPool.clear()
     bufferedObjectOperations.clear()
+    emitterScope.cancel("ObjectsManager disposed")
   }
 }

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

@@ -0,0 +1,33 @@
+package io.ably.lib.objects
+
+import io.ably.lib.util.EventEmitter
+
+/**
+ * @spec RTO2 - enum representing objects state
+ */
+internal enum class ObjectsState {
+  INITIALIZED,
+  SYNCING,
+  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
+)
+
+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?) {
+    listener?.onStateChanged(event!!)
+  }
+}