[ECO-5426] Added spec annotation comments for LiveMap, LiveCounter and LiveObjects

Author: sacOO7

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

@@ -44,9 +44,9 @@ internal class DefaultLiveObjects(private val channelName: String, private val a
   private val syncObjectsDataPool = ConcurrentHashMap<String, ObjectState>()
   private var currentSyncId: String? = null
   /**
-   * @spec RTO5 - Buffered object operations during sync
+   * @spec RTO7 - Buffered object operations during sync
    */
-  private val bufferedObjectOperations = mutableListOf<ObjectMessage>()
+  private val bufferedObjectOperations = mutableListOf<ObjectMessage>() // RTO7a
 
   /**
    * @spec RTO1 - Returns the root LiveMap object with proper validation and sync waiting
@@ -130,18 +130,21 @@ internal class DefaultLiveObjects(private val channelName: String, private val a
   /**
    * Handles object messages (non-sync messages).
    *
-   * @spec RTO5 - Buffers messages if not synced, applies immediately if synced
+   * @spec RTO8 - Buffers messages if not synced, applies immediately if synced
    */
   private fun handleObjectMessages(objectMessages: List<ObjectMessage>) {
     if (state != ObjectsState.SYNCED) {
-      // Buffer messages if not synced yet
+      // RTO7 - The client receives object messages in realtime over the channel concurrently with the sync sequence.
+      // Some of the incoming object messages may have already been applied to the objects described in
+      // the sync sequence, but others may not; therefore we must buffer these messages so that we can apply
+      // them to the objects once the sync is complete.
       Log.v(tag, "Buffering ${objectMessages.size} object messages, state: $state")
-      bufferedObjectOperations.addAll(objectMessages)
+      bufferedObjectOperations.addAll(objectMessages) // RTO8a
       return
     }
 
     // Apply messages immediately if synced
-    applyObjectMessages(objectMessages)
+    applyObjectMessages(objectMessages) // RTO8b
   }
 
   /**
@@ -154,7 +157,7 @@ internal class DefaultLiveObjects(private val channelName: String, private val a
     val newSyncSequence = currentSyncId != syncId
     if (newSyncSequence) {
       // RTO5a2 - new sync sequence started
-      startNewSync(syncId) // RTO5a2a
+      startNewSync(syncId)
     }
 
     // RTO5a3 - continue current sync sequence
@@ -198,8 +201,8 @@ internal class DefaultLiveObjects(private val channelName: String, private val a
     Log.v(tag, "Starting new sync sequence: syncId=$syncId")
 
     // need to discard all buffered object operation messages on new sync start
-    bufferedObjectOperations.clear()
-    syncObjectsDataPool.clear()
+    bufferedObjectOperations.clear() // RTO5a2b
+    syncObjectsDataPool.clear() // RTO5a2a
     currentSyncId = syncId
     stateChange(ObjectsState.SYNCING, false)
   }
@@ -214,9 +217,9 @@ internal class DefaultLiveObjects(private val channelName: String, private val a
     applySync()
     // should apply buffered object operations after we applied the sync.
     // can use regular non-sync object.operation logic
-    applyObjectMessages(bufferedObjectOperations)
+    applyObjectMessages(bufferedObjectOperations) // RTO5c6
 
-    bufferedObjectOperations.clear()
+    bufferedObjectOperations.clear() // RTO5c5
     syncObjectsDataPool.clear() // RTO5c4
     currentSyncId = null // RTO5c3
     stateChange(ObjectsState.SYNCED, deferStateEvent)
@@ -264,19 +267,26 @@ internal class DefaultLiveObjects(private val channelName: String, private val a
   /**
    * Applies object messages to objects.
    *
-   * @spec RTO6 - Creates zero-value objects if they don't exist
+   * @spec RTO9 - Creates zero-value objects if they don't exist
    */
   private fun applyObjectMessages(objectMessages: List<ObjectMessage>) {
+    // RTO9a
     for (objectMessage in objectMessages) {
       if (objectMessage.operation == null) {
+        // RTO9a1
         Log.w(tag, "Object message received without operation field, skipping message: ${objectMessage.id}")
         continue
       }
 
-      val objectOperation: ObjectOperation = objectMessage.operation
-      // RTO6a - get or create the zero value object in the pool
-      val obj = objectsPool.createZeroValueObjectIfNotExists(objectOperation.objectId)
-      obj.applyOperation(objectOperation, objectMessage)
+      val objectOperation: ObjectOperation = objectMessage.operation // RTO9a2
+      // RTO9a2a - we can receive an op for an object id we don't have yet in the pool. instead of buffering such operations,
+      // we can create a zero-value object for the provided object id and apply the operation to that zero-value object.
+      // this also means that all objects are capable of applying the corresponding *_CREATE ops on themselves,
+      // since they need to be able to eventually initialize themselves from that *_CREATE op.
+      // so to simplify operations handling, we always try to create a zero-value object in the pool first,
+      // and then we can always apply the operation on the existing object in the pool.
+      val obj = objectsPool.createZeroValueObjectIfNotExists(objectOperation.objectId) // RTO9a2a1
+      obj.applyOperation(objectOperation, objectMessage) // RTO9a2a2, RTO9a2a3
     }
   }
 

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

@@ -12,26 +12,23 @@ import io.ably.lib.util.Log
  * Base implementation of LiveObject interface.
  * Provides common functionality for all live objects.
  *
- * @spec RTLM1/RTLC1 - Base class for LiveMap/LiveCounter objects
+ * @spec RTLO1/RTLO2 - Base class for LiveMap/LiveCounter objects
  */
 internal abstract class BaseLiveObject(
-  protected val objectId: String,
+  protected val objectId: String, // // RTLO3a
   protected val adapter: LiveObjectsAdapter
 ) {
 
   protected open val tag = "BaseLiveObject"
   internal var isTombstoned = false
   internal var tombstonedAt: Long? = null
 
-  /**
-   * @spec RTLM6/RTLC6 - Map of serials keyed by site code for LiveMap/LiveCounter
-   */
-  protected val siteTimeserials = mutableMapOf<String, String>()
+  protected val siteTimeserials = mutableMapOf<String, String>() // RTLO3b
 
   /**
-   * @spec RTLM6/RTLC6 - Flag to track if create operation has been merged for LiveMap/LiveCounter
+   * @spec RTLO3 - Flag to track if create operation has been merged for LiveMap/LiveCounter
    */
-  protected var createOperationIsMerged = false
+  protected var createOperationIsMerged = false // RTLO3c
 
   fun notifyUpdated(update: Any) {
     // TODO: Implement event emission for updates
@@ -41,17 +38,17 @@ internal abstract class BaseLiveObject(
   /**
    * Checks if an operation can be applied based on serial comparison.
    *
-   * @spec RTLM9/RTLC9 - Serial comparison logic for LiveMap/LiveCounter operations
+   * @spec RTLO4a - Serial comparison logic for LiveMap/LiveCounter operations
    */
   protected fun canApplyOperation(siteCode: String?, serial: String?): Boolean {
     if (serial.isNullOrEmpty()) {
-      throw objectError("Invalid serial: $serial")
+      throw objectError("Invalid serial: $serial") // RTLO4a3
     }
     if (siteCode.isNullOrEmpty()) {
-      throw objectError("Invalid site code: $siteCode")
+      throw objectError("Invalid site code: $siteCode") // RTLO4a3
     }
-    val existingSiteSerial = siteTimeserials[siteCode]
-    return existingSiteSerial == null || serial > existingSiteSerial
+    val existingSiteSerial = siteTimeserials[siteCode] // RTLO4a4
+    return existingSiteSerial == null || serial > existingSiteSerial // RTLO4a5, RTLO4a6
   }
 
   /**

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

@@ -8,7 +8,6 @@ import io.ably.lib.objects.ObjectMessage
 import io.ably.lib.objects.ObjectOperation
 import io.ably.lib.objects.ObjectOperationAction
 import io.ably.lib.objects.ObjectState
-import io.ably.lib.objects.ObjectsPool
 import io.ably.lib.objects.ablyException
 import io.ably.lib.objects.objectError
 import io.ably.lib.types.AblyException
@@ -23,7 +22,7 @@ import io.ably.lib.util.Log
 internal class DefaultLiveCounter(
   objectId: String,
   adapter: LiveObjectsAdapter,
-) : BaseLiveObject(objectId, adapter), LiveCounter {
+) : LiveCounter, BaseLiveObject(objectId, adapter) {
 
   override val tag = "LiveCounter"
   /**
@@ -86,6 +85,7 @@ internal class DefaultLiveCounter(
     val opSiteCode = message.siteCode
 
     if (!canApplyOperation(opSiteCode, opSerial)) {
+      // RTLC7b
       Log.v(
         tag,
         "Skipping ${operation.action} op: op serial $opSerial <= site serial ${siteTimeserials[opSiteCode]}; objectId=$objectId"
@@ -94,24 +94,24 @@ internal class DefaultLiveCounter(
     }
     // should update stored site serial immediately. doesn't matter if we successfully apply the op,
     // as it's important to mark that the op was processed by the object
-    updateTimeSerial(opSiteCode!!, opSerial!!)
+    updateTimeSerial(opSiteCode!!, opSerial!!) // RTLC7c
 
     if (isTombstoned) {
       // this object is tombstoned so the operation cannot be applied
       return;
     }
 
     val update = when (operation.action) {
-      ObjectOperationAction.CounterCreate -> applyCounterCreate(operation)
+      ObjectOperationAction.CounterCreate -> applyCounterCreate(operation) // RTLC7d1
       ObjectOperationAction.CounterInc -> {
         if (operation.counterOp != null) {
-          applyCounterInc(operation.counterOp)
+          applyCounterInc(operation.counterOp) // RTLC7d2
         } else {
           throw payloadError(operation)
         }
       }
       ObjectOperationAction.ObjectDelete -> applyObjectDelete()
-      else -> throw objectError("Invalid ${operation.action} op for LiveCounter objectId=${objectId}")
+      else -> throw objectError("Invalid ${operation.action} op for LiveCounter objectId=${objectId}") // RTLC7d3
     }
 
     notifyUpdated(update)
@@ -124,40 +124,44 @@ internal class DefaultLiveCounter(
   }
 
   /**
-   * @spec RTLC6d - Merges initial data from create operation
+   * @spec RTLC8 - Applies counter create operation
    */
   private fun applyCounterCreate(operation: ObjectOperation): Map<String, Long> {
     if (createOperationIsMerged) {
+      // RTLC8b
+      // There can't be two different create operation for the same object id, because the object id
+      // fully encodes that operation. This means we can safely ignore any new incoming create operations
+      // if we already merged it once.
       Log.v(
         tag,
         "Skipping applying COUNTER_CREATE op on a counter instance as it was already applied before; objectId=$objectId"
       )
       return mapOf()
     }
 
-    return mergeInitialDataFromCreateOperation(operation)
+    return mergeInitialDataFromCreateOperation(operation) // RTLC8c
   }
 
   /**
-   * @spec RTLC8 - Applies counter increment operation
+   * @spec RTLC9 - Applies counter increment operation
    */
   private fun applyCounterInc(counterOp: ObjectCounterOp): Map<String, Long> {
     val amount = counterOp.amount?.toLong() ?: 0
-    data += amount
+    data += amount // RTLC9b
     return mapOf("amount" to amount)
   }
 
   /**
-   * @spec RTLC6d - Merges initial data from create operation
+   * @spec RTLC10 - Merges initial data from create operation
    */
   private fun mergeInitialDataFromCreateOperation(operation: ObjectOperation): Map<String, Long> {
     // 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,
     // which we're going to add now.
     val count = operation.counter?.count?.toLong() ?: 0
-    data += count // RTLC6d1
-    createOperationIsMerged = true // RTLC6d2
+    data += count // RTLC10a
+    createOperationIsMerged = true // RTLC10b
     return mapOf("amount" to count)
   }
 
@@ -186,8 +190,12 @@ internal class DefaultLiveCounter(
     TODO("Not yet implemented")
   }
 
+  /**
+   * @spec RTLC5 - Returns the current counter value
+   */
   override fun value(): Long {
-    TODO("Not yet implemented")
+    // RTLC5a, RTLC5b - Configuration validation would be done here
+    return data // RTLC5c
   }
 
   companion object {