[ECO-5447] Updated ObjectMessage with server provided serialTimestamp

1. Updated MsgpackSerialization accordingly.
2. Updated LiveMapManagerTests, added few more tests related to changes

Author: sacOO7

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

@@ -125,6 +125,12 @@ internal data class ObjectMapEntry(
    */
   val timeserial: String? = null,
 
+  /**
+   * A timestamp from the [timeserial] field. Only present if [tombstone] is `true`
+   * Spec: OME2d
+   */
+  val serialTimestamp: Long? = null,
+
   /**
    * The data that represents the value of the map entry.
    * Spec: OME2c
@@ -336,6 +342,12 @@ internal data class ObjectMessage(
    */
   val serial: String? = null,
 
+  /**
+   * A timestamp from the [serial] field.
+   * Spec: OM2j
+   */
+  val serialTimestamp: Long? = null,
+
   /**
    * An opaque string used as a key to update the map of serial values on an object.
    * Spec: OM2i

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

@@ -38,6 +38,7 @@ internal fun ObjectMessage.writeMsgpack(packer: MessagePacker) {
   if (operation != null) fieldCount++
   if (objectState != null) fieldCount++
   if (serial != null) fieldCount++
+  if (serialTimestamp != null) fieldCount++
   if (siteCode != null) fieldCount++
 
   packer.packMapHeader(fieldCount)
@@ -82,6 +83,11 @@ internal fun ObjectMessage.writeMsgpack(packer: MessagePacker) {
     packer.packString(serial)
   }
 
+  if (serialTimestamp != null) {
+    packer.packString("serialTimestamp")
+    packer.packLong(serialTimestamp)
+  }
+
   if (siteCode != null) {
     packer.packString("siteCode")
     packer.packString(siteCode)
@@ -107,6 +113,7 @@ internal fun readObjectMessage(unpacker: MessageUnpacker): ObjectMessage {
   var operation: ObjectOperation? = null
   var objectState: ObjectState? = null
   var serial: String? = null
+  var serialTimestamp: Long? = null
   var siteCode: String? = null
 
   for (i in 0 until fieldCount) {
@@ -127,6 +134,7 @@ internal fun readObjectMessage(unpacker: MessageUnpacker): ObjectMessage {
       "operation" -> operation = readObjectOperation(unpacker)
       "object" -> objectState = readObjectState(unpacker)
       "serial" -> serial = unpacker.unpackString()
+      "serialTimestamp" -> serialTimestamp = unpacker.unpackLong()
       "siteCode" -> siteCode = unpacker.unpackString()
       else -> unpacker.skipValue()
     }
@@ -141,6 +149,7 @@ internal fun readObjectMessage(unpacker: MessageUnpacker): ObjectMessage {
     operation = operation,
     objectState = objectState,
     serial = serial,
+    serialTimestamp = serialTimestamp,
     siteCode = siteCode
   )
 }
@@ -558,6 +567,7 @@ private fun ObjectMapEntry.writeMsgpack(packer: MessagePacker) {
 
   if (tombstone != null) fieldCount++
   if (timeserial != null) fieldCount++
+  if (serialTimestamp != null) fieldCount++
   if (data != null) fieldCount++
 
   packer.packMapHeader(fieldCount)
@@ -572,6 +582,11 @@ private fun ObjectMapEntry.writeMsgpack(packer: MessagePacker) {
     packer.packString(timeserial)
   }
 
+  if (serialTimestamp != null) {
+    packer.packString("serialTimestamp")
+    packer.packLong(serialTimestamp)
+  }
+
   if (data != null) {
     packer.packString("data")
     data.writeMsgpack(packer)
@@ -586,6 +601,7 @@ private fun readObjectMapEntry(unpacker: MessageUnpacker): ObjectMapEntry {
 
   var tombstone: Boolean? = null
   var timeserial: String? = null
+  var serialTimestamp: Long? = null
   var data: ObjectData? = null
 
   for (i in 0 until fieldCount) {
@@ -600,12 +616,13 @@ private fun readObjectMapEntry(unpacker: MessageUnpacker): ObjectMapEntry {
     when (fieldName) {
       "tombstone" -> tombstone = unpacker.unpackBoolean()
       "timeserial" -> timeserial = unpacker.unpackString()
+      "serialTimestamp" -> serialTimestamp = unpacker.unpackLong()
       "data" -> data = readObjectData(unpacker)
       else -> unpacker.skipValue()
     }
   }
 
-  return ObjectMapEntry(tombstone = tombstone, timeserial = timeserial, data = data)
+  return ObjectMapEntry(tombstone = tombstone, timeserial = timeserial, serialTimestamp = serialTimestamp, data = data)
 }
 
 /**

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

@@ -124,7 +124,7 @@ internal class DefaultLiveMap private constructor(
   }
 
   override fun applyObjectOperation(operation: ObjectOperation, message: ObjectMessage) {
-    liveMapManager.applyOperation(operation, message.serial)
+    liveMapManager.applyOperation(operation, message.serial, message.serialTimestamp)
   }
 
   override fun clearData(): LiveMapUpdate {

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

@@ -33,7 +33,7 @@ internal class LiveMapManager(private val liveMap: DefaultLiveMap): LiveMapChang
       objectState.map?.entries?.forEach { (key, entry) ->
         liveMap.data[key] = LiveMapEntry(
           isTombstoned = entry.tombstone ?: false,
-          tombstonedAt = if (entry.tombstone == true) System.currentTimeMillis() else null,
+          tombstonedAt = if (entry.tombstone == true) entry.serialTimestamp ?: System.currentTimeMillis() else null,
           timeserial = entry.timeserial,
           data = entry.data
         )
@@ -51,19 +51,19 @@ internal class LiveMapManager(private val liveMap: DefaultLiveMap): LiveMapChang
   /**
    * @spec RTLM15 - Applies operations to LiveMap
    */
-  internal fun applyOperation(operation: ObjectOperation, messageTimeserial: String?) {
+  internal fun applyOperation(operation: ObjectOperation, serial: String?, serialTimestamp: Long?) {
     val update = when (operation.action) {
       ObjectOperationAction.MapCreate -> applyMapCreate(operation) // RTLM15d1
       ObjectOperationAction.MapSet -> {
         if (operation.mapOp != null) {
-          applyMapSet(operation.mapOp, messageTimeserial) // RTLM15d2
+          applyMapSet(operation.mapOp, serial) // RTLM15d2
         } else {
           throw objectError("No payload found for ${operation.action} op for LiveMap objectId=${objectId}")
         }
       }
       ObjectOperationAction.MapRemove -> {
         if (operation.mapOp != null) {
-          applyMapRemove(operation.mapOp, messageTimeserial) // RTLM15d3
+          applyMapRemove(operation.mapOp, serial, serialTimestamp) // RTLM15d3
         } else {
           throw objectError("No payload found for ${operation.action} op for LiveMap objectId=${objectId}")
         }
@@ -132,7 +132,6 @@ internal class LiveMapManager(private val liveMap: DefaultLiveMap): LiveMapChang
       // RTLM7a2 - Replace existing entry with new one instead of mutating
       liveMap.data[mapOp.key] = LiveMapEntry(
         isTombstoned = false, // RTLM7a2c
-        tombstonedAt = null,
         timeserial = timeSerial, // RTLM7a2b
         data = mapOp.data // RTLM7a2a
       )
@@ -154,6 +153,7 @@ internal class LiveMapManager(private val liveMap: DefaultLiveMap): LiveMapChang
   private fun applyMapRemove(
     mapOp: ObjectMapOp, // RTLM8c1
     timeSerial: String?, // RTLM8c2
+    timeStamp: Long?, // RTLM8c3
   ): LiveMapUpdate {
     val existingEntry = liveMap.data[mapOp.key]
 
@@ -168,19 +168,28 @@ internal class LiveMapManager(private val liveMap: DefaultLiveMap): LiveMapChang
       return noOpMapUpdate
     }
 
+    val tombstonedAt = if (timeStamp != null) timeStamp else {
+      Log.w(
+        tag,
+        "No timestamp provided for MAP_REMOVE op on key=\"${mapOp.key}\"; using current time as tombstone time; " +
+          "objectId=${objectId}"
+      )
+      System.currentTimeMillis()
+    }
+
     if (existingEntry != null) {
       // RTLM8a2 - Replace existing entry with new one instead of mutating
       liveMap.data[mapOp.key] = LiveMapEntry(
         isTombstoned = true, // RTLM8a2c
-        tombstonedAt = System.currentTimeMillis(),
+        tombstonedAt = tombstonedAt,
         timeserial = timeSerial, // RTLM8a2b
         data = null // RTLM8a2a
       )
     } else {
       // RTLM8b, RTLM8b1
       liveMap.data[mapOp.key] = LiveMapEntry(
         isTombstoned = true, // RTLM8b2
-        tombstonedAt = System.currentTimeMillis(),
+        tombstonedAt = tombstonedAt,
         timeserial = timeSerial
       )
     }
@@ -224,7 +233,7 @@ internal class LiveMapManager(private val liveMap: DefaultLiveMap): LiveMapChang
       val opTimeserial = entry.timeserial
       val update = if (entry.tombstone == true) {
         // RTLM17a2 - entry in MAP_CREATE op is removed, try to apply MAP_REMOVE op
-        applyMapRemove(ObjectMapOp(key), opTimeserial)
+        applyMapRemove(ObjectMapOp(key), opTimeserial, entry.serialTimestamp)
       } else {
         // RTLM17a1 - entry in MAP_CREATE op is not removed, try to set it via MAP_SET op
         applyMapSet(ObjectMapOp(key, entry.data), opTimeserial)