[ECO-5482] Implemented createMap and createCounter methods for objects

Author: sacOO7

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

@@ -15,6 +15,7 @@
 public class Adapter implements LiveObjectsAdapter {
     private final AblyRealtime ably;
     private static final String TAG = LiveObjectsAdapter.class.getName();
+    private volatile Long serverTimeOffset = null;
 
     public Adapter(@NotNull AblyRealtime ably) {
         this.ably = ably;
@@ -77,4 +78,18 @@ public ChannelState getChannelState(@NotNull String channelName) {
     public @NotNull ConnectionManager getConnectionManager() {
         return ably.connection.connectionManager;
     }
+
+    @Override
+    public long getServerTime() throws AblyException {
+        if (serverTimeOffset == null) {
+            synchronized (this) {
+                if (serverTimeOffset == null) { // Double-checked locking to ensure thread safety
+                    long serverTime = ably.time();
+                    serverTimeOffset = serverTime - System.currentTimeMillis();
+                    return serverTime;
+                }
+            }
+        }
+        return System.currentTimeMillis() + serverTimeOffset;
+    }
 }

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

@@ -7,6 +7,7 @@
 import io.ably.lib.types.ChannelMode;
 import io.ably.lib.types.ClientOptions;
 import io.ably.lib.types.ProtocolMessage;
+import org.jetbrains.annotations.Blocking;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -73,5 +74,14 @@ public interface LiveObjectsAdapter {
      * @return the connection manager instance
      */
     @NotNull ConnectionManager getConnectionManager();
+
+    /**
+     * Retrieves the current time in milliseconds from the Ably server.
+     * On first call, queries the server time and caches the offset from local time.
+     * Subsequent calls return the local time adjusted by this offset.
+     * Spec: RTO16
+     */
+    @Blocking
+    long getServerTime() throws AblyException;
 }
 

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

@@ -1,8 +1,12 @@
 package io.ably.lib.objects
 
+import io.ably.lib.objects.serialization.gson
 import io.ably.lib.objects.state.ObjectsStateChange
 import io.ably.lib.objects.state.ObjectsStateEvent
+import io.ably.lib.objects.type.ObjectType
 import io.ably.lib.objects.type.counter.LiveCounter
+import io.ably.lib.objects.type.livecounter.DefaultLiveCounter
+import io.ably.lib.objects.type.livemap.DefaultLiveMap
 import io.ably.lib.objects.type.map.LiveMap
 import io.ably.lib.objects.type.map.LiveMapValue
 import io.ably.lib.realtime.ChannelState
@@ -57,40 +61,28 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
 
   override fun getRoot(): LiveMap = runBlocking { getRootAsync() }
 
-  override fun createMap(): LiveMap {
-     return createMap(mutableMapOf())
-  }
+  override fun createMap(): LiveMap = createMap(mutableMapOf())
 
-  override fun createMap(entries: MutableMap<String, LiveMapValue>): LiveMap {
-    TODO("Not yet implemented")
-  }
+  override fun createMap(entries: MutableMap<String, LiveMapValue>): LiveMap = runBlocking { createMapAsync(entries) }
 
-  override fun createCounter(): LiveCounter {
-    return createCounter(0)
-  }
+  override fun createCounter(): LiveCounter = createCounter(0)
 
-  override fun createCounter(initialValue: Number): LiveCounter {
-    TODO("Not yet implemented")
-  }
+  override fun createCounter(initialValue: Number): LiveCounter = runBlocking { createCounterAsync(initialValue) }
 
   override fun getRootAsync(callback: Callback<LiveMap>) {
     asyncScope.launchWithCallback(callback) { getRootAsync() }
   }
 
-  override fun createMapAsync(callback: Callback<LiveMap>) {
-    TODO("Not yet implemented")
-  }
+  override fun createMapAsync(callback: Callback<LiveMap>) = createMapAsync(mutableMapOf(), callback)
 
   override fun createMapAsync(entries: MutableMap<String, LiveMapValue>, callback: Callback<LiveMap>) {
-    TODO("Not yet implemented")
+    asyncScope.launchWithCallback(callback) { createMapAsync(entries) }
   }
 
-  override fun createCounterAsync(callback: Callback<LiveCounter>) {
-    TODO("Not yet implemented")
-  }
+  override fun createCounterAsync(callback: Callback<LiveCounter>) = createCounterAsync(0, callback)
 
   override fun createCounterAsync(initialValue: Number, callback: Callback<LiveCounter>) {
-    TODO("Not yet implemented")
+    asyncScope.launchWithCallback(callback) { createCounterAsync(initialValue) }
   }
 
   override fun on(event: ObjectsStateEvent, listener: ObjectsStateChange.Listener): ObjectsSubscription =
@@ -106,6 +98,81 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
     objectsPool.get(ROOT_OBJECT_ID) as LiveMap
   }
 
+  private suspend fun createMapAsync(entries: MutableMap<String, LiveMapValue>): LiveMap {
+    adapter.throwIfInvalidWriteApiConfiguration(channelName)
+
+    // Create initial value operation
+    val initialMapValue = DefaultLiveMap.createInitialValue(entries)
+
+    // Create initial value JSON string
+    val initialValueJSONString = gson.toJson(initialMapValue)
+
+    // Create object ID from initial value
+    val (objectId, nonce) = fromInitialValue(ObjectType.Map, initialValueJSONString)
+
+    // Create ObjectMessage with the operation
+    val msg = ObjectMessage(
+      operation = ObjectOperation(
+        action = ObjectOperationAction.MapCreate,
+        objectId = objectId,
+        map = initialMapValue.map,
+        nonce = nonce,
+        initialValue = initialValueJSONString,
+      )
+    )
+
+    // Publish the message
+    publish(arrayOf(msg))
+
+    // Check if object already exists in pool (from echoed message)
+    return objectsPool.get(objectId) as? LiveMap ?: withContext(sequentialScope.coroutineContext) {
+      objectsPool.createZeroValueObjectIfNotExists(objectId) as LiveMap
+    }
+  }
+
+  private suspend fun createCounterAsync(initialValue: Number): LiveCounter {
+    adapter.throwIfInvalidWriteApiConfiguration(channelName)
+
+    // Validate input parameter
+    if (initialValue.toDouble().isNaN() || initialValue.toDouble().isInfinite()) {
+      throw objectError("Counter value should be a valid number")
+    }
+
+    val initialCounterValue = DefaultLiveCounter.createInitialValue(initialValue)
+    // Create initial value operation
+    val initialValueJSONString = gson.toJson(initialCounterValue)
+
+    // Create object ID from initial value
+    val (objectId, nonce) = fromInitialValue(ObjectType.Counter, initialValueJSONString)
+
+    // Create ObjectMessage with the operation
+    val msg = ObjectMessage(
+      operation = ObjectOperation(
+        action = ObjectOperationAction.CounterCreate,
+        objectId = objectId,
+        counter = initialCounterValue.counter,
+        nonce = nonce,
+        initialValue = initialValueJSONString
+      )
+    )
+
+    // Publish the message
+    publish(arrayOf(msg))
+
+    // Check if object already exists in pool (from echoed message)
+    return objectsPool.get(objectId) as? LiveCounter ?: withContext(sequentialScope.coroutineContext) {
+      objectsPool.createZeroValueObjectIfNotExists(objectId) as LiveCounter
+    }
+  }
+
+  private suspend fun fromInitialValue(objectType: ObjectType, initialValue: String): Pair<String, String> {
+    val nonce = generateNonce()
+    val msTimestamp = withContext(Dispatchers.IO) {
+      adapter.getServerTime()
+    }
+    return Pair(ObjectId.fromInitialValue(objectType, initialValue, nonce, msTimestamp).toString(), nonce)
+  }
+
   /**
    * Spec: RTO15
    */

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

@@ -1,5 +1,7 @@
 package io.ably.lib.objects
 
+import io.ably.lib.objects.type.BaseLiveObject
+import io.ably.lib.objects.type.map.LiveMapValue
 import io.ably.lib.realtime.ChannelState
 import io.ably.lib.realtime.CompletionListener
 import io.ably.lib.types.ChannelMode
@@ -103,3 +105,18 @@ internal class Binary(val data: ByteArray) {
 internal fun Binary.size(): Int {
   return data.size
 }
+
+internal data class CounterCreatePayload(
+  val counter: ObjectCounter
+)
+
+internal data class MapCreatePayload(
+  val map: ObjectMap
+)
+
+internal fun fromLiveMapValue(value: LiveMapValue) : ObjectData {
+  return when {
+    value.isLiveMap || value.isLiveCounter -> ObjectData(objectId = (value.value as BaseLiveObject).objectId)
+    else -> ObjectData(value = ObjectValue(value.value))
+  }
+}

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

@@ -1,6 +1,9 @@
 package io.ably.lib.objects
 
 import io.ably.lib.objects.type.ObjectType
+import java.nio.charset.StandardCharsets
+import java.security.MessageDigest
+import java.util.Base64
 
 internal class ObjectId private constructor(
   internal val type: ObjectType,
@@ -15,10 +18,19 @@ internal class ObjectId private constructor(
   }
 
   companion object {
+
+    internal fun fromInitialValue(objectType: ObjectType, initialValue: String, nonce: String, msTimeStamp: Long): ObjectId {
+      val valueForHash = "$initialValue:$nonce".toByteArray(StandardCharsets.UTF_8)
+      val hashBytes = MessageDigest.getInstance("SHA-256").digest(valueForHash)
+      val urlSafeHash = Base64.getUrlEncoder().withoutPadding().encodeToString(hashBytes)
+
+      return ObjectId(objectType, urlSafeHash, msTimeStamp)
+    }
+
     /**
      * Creates ObjectId instance from hashed object id string.
      */
-    fun fromString(objectId: String): ObjectId {
+    internal fun fromString(objectId: String): ObjectId {
       if (objectId.isEmpty()) {
         throw objectError("Invalid object id: $objectId")
       }

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

@@ -5,6 +5,7 @@ import io.ably.lib.types.Callback
 import io.ably.lib.types.ErrorInfo
 import io.ably.lib.util.Log
 import kotlinx.coroutines.*
+import java.nio.charset.StandardCharsets
 import java.util.concurrent.CancellationException
 
 internal fun ablyException(
@@ -47,7 +48,7 @@ internal fun objectError(errorMessage: String, cause: Throwable? = null): AblyEx
  * e.g. "Hello" has a byte size of 5, while "你" has a byte size of 3 and "😊" has a byte size of 4.
  */
 internal val String.byteSize: Int
-  get() = this.toByteArray(Charsets.UTF_8).size
+  get() = this.toByteArray(StandardCharsets.UTF_8).size
 
 /**
  * A channel-specific coroutine scope for executing callbacks asynchronously in the LiveObjects system.
@@ -78,3 +79,11 @@ internal class ObjectsAsyncScope(channelName: String) {
     scope.coroutineContext.cancelChildren(cause)
   }
 }
+
+/**
+ * Generates a random nonce string for object creation.
+ */
+internal fun generateNonce(): String {
+  val chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
+  return (1..16).map { chars.random() }.joinToString("")
+}

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

@@ -105,5 +105,14 @@ internal class DefaultLiveCounter private constructor(
     internal fun zeroValue(objectId: String, liveObjects: DefaultLiveObjects): DefaultLiveCounter {
       return DefaultLiveCounter(objectId, liveObjects)
     }
+
+    /**
+     * Creates initial value operation for counter creation.
+     */
+    internal fun createInitialValue(count: Number): CounterCreatePayload {
+      return CounterCreatePayload(
+        counter = ObjectCounter(count = count.toDouble())
+      )
+    }
   }
 }

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

@@ -153,5 +153,24 @@ internal class DefaultLiveMap private constructor(
     internal fun zeroValue(objectId: String, objects: DefaultLiveObjects): DefaultLiveMap {
       return DefaultLiveMap(objectId, objects)
     }
+
+    /**
+     * Creates an ObjectMap from map entries.
+     */
+    internal fun createInitialValue(entries: MutableMap<String, LiveMapValue>): MapCreatePayload {
+      val mapEntries = entries.mapValues { (_, value) ->
+        ObjectMapEntry(
+          tombstone = false,
+          data = fromLiveMapValue(value)
+        )
+      }
+
+      return MapCreatePayload(
+        map = ObjectMap(
+          semantics = MapSemantics.LWW,
+          entries = mapEntries
+        )
+      )
+    }
   }
 }

live-objects/src/test/kotlin/io/ably/lib/objects/unit/ObjectIdTest.kt

@@ -52,4 +52,24 @@ class ObjectIdTest {
     assertEquals(92_000, exception.errorInfo?.code)
     assertEquals(500, exception.errorInfo?.statusCode)
   }
+
+  @Test
+  fun testFromInitialValue() {
+    val objectType = ObjectType.Map
+    val initialValue = "test-value"
+    val nonce = "test-nonce"
+    val msTimestamp = 1640995200000L
+
+    val objectId = ObjectId.fromInitialValue(objectType, initialValue, nonce, msTimestamp)
+    // Verify the string format follows the expected pattern: type:hash@timestamp
+    val objectIdString = objectId.toString()
+    assertTrue(objectIdString.startsWith("map:"))
+    assertTrue(objectIdString.contains("@"))
+    assertTrue(objectIdString.endsWith(msTimestamp.toString()))
+
+    val expectedHash = "GSjv-adTaJPL8-382qF3JuIyE4TCc6QKIIqb577pz00"
+    // Verify the hash value matches expected
+    val hashPart = objectIdString.substring(4, objectIdString.indexOf("@"))
+    assertEquals(expectedHash, hashPart)
+  }
 }

live-objects/src/test/kotlin/io/ably/lib/objects/unit/ObjectMessageSizeTest.kt

@@ -168,7 +168,7 @@ class ObjectMessageSizeTest {
     }
     // Assert on error code and message
     assertEquals(40009, exception.errorInfo.code)
-    val expectedMessage = "ObjectMessage size 66560 exceeds maximum allowed size of 65536 bytes"
+    val expectedMessage = "ObjectMessages size 66560 exceeds maximum allowed size of 65536 bytes"
     assertEquals(expectedMessage, exception.errorInfo.message)
   }
 }