[ECO-5482] Implemented realtime write methods for LiveMap and LiveCounter

- Added integration test performing counter ops using realtime write API
- Added integration test performing map ops using realtime write API

Author: sacOO7

lib/src/main/java/io/ably/lib/objects/type/map/LiveMap.java

@@ -112,7 +112,7 @@ public interface LiveMap extends LiveMapChange {
      * @param callback the callback to handle the result or any errors.
      */
     @NonBlocking
-    void setAsync(@NotNull String keyName, @NotNull Object value, @NotNull Callback<Void> callback);
+    void setAsync(@NotNull String keyName, @NotNull LiveMapValue value, @NotNull Callback<Void> callback);
 
     /**
      * Asynchronously removes the specified key and its associated value from the map.

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

@@ -53,7 +53,7 @@ internal class DefaultLiveObjects(internal val channelName: String, internal val
   /**
    * Provides a channel-specific scope for safely executing asynchronous operations with callbacks.
    */
-  private val asyncScope = ObjectsAsyncScope(channelName)
+  internal val asyncScope = ObjectsAsyncScope(channelName)
 
   init {
     incomingObjectsHandler = initializeHandlerForIncomingObjectMessages()

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

@@ -1,7 +1,5 @@
 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
@@ -113,10 +111,3 @@ internal data class CounterCreatePayload(
 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/Utils.kt

@@ -75,6 +75,20 @@ internal class ObjectsAsyncScope(channelName: String) {
     }
   }
 
+  internal fun launchWithVoidCallback(callback: Callback<Void>, block: suspend () -> Unit) {
+    scope.launch {
+      try {
+        block()
+        try { callback.onSuccess(null) } catch (t: Throwable) {
+          Log.e(tag, "Error occurred while executing callback's onSuccess handler", t)
+        } // catch and don't rethrow error from callback
+      } catch (throwable: Throwable) {
+        val exception = throwable as? AblyException
+        callback.onError(exception?.errorInfo)
+      }
+    }
+  }
+
   internal fun cancel(cause: CancellationException) {
     scope.coroutineContext.cancelChildren(cause)
   }

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

@@ -13,6 +13,7 @@ import io.ably.lib.objects.type.noOp
 import io.ably.lib.types.Callback
 import java.util.concurrent.atomic.AtomicReference
 import io.ably.lib.util.Log
+import kotlinx.coroutines.runBlocking
 
 /**
  * Implementation of LiveObject for LiveCounter.
@@ -39,21 +40,18 @@ internal class DefaultLiveCounter private constructor(
 
   private val channelName = liveObjects.channelName
   private val adapter: LiveObjectsAdapter get() = liveObjects.adapter
+  private val asyncScope get() = liveObjects.asyncScope
 
-  override fun increment(amount: Number) {
-    TODO("Not yet implemented")
-  }
+  override fun increment(amount: Number) = runBlocking { incrementAsync(amount.toDouble()) }
 
-  override fun decrement(amount: Number) {
-    TODO("Not yet implemented")
-  }
+  override fun decrement(amount: Number) = runBlocking { incrementAsync(-amount.toDouble()) }
 
   override fun incrementAsync(amount: Number, callback: Callback<Void>) {
-    TODO("Not yet implemented")
+    asyncScope.launchWithVoidCallback(callback) { incrementAsync(amount.toDouble()) }
   }
 
   override fun decrementAsync(amount: Number, callback: Callback<Void>) {
-    TODO("Not yet implemented")
+    asyncScope.launchWithVoidCallback(callback) { incrementAsync(-amount.toDouble()) }
   }
 
   override fun value(): Double {
@@ -72,6 +70,28 @@ internal class DefaultLiveCounter private constructor(
 
   override fun validate(state: ObjectState) = liveCounterManager.validate(state)
 
+  private suspend fun incrementAsync(amount: Double) {
+    // Validate write API configuration
+    adapter.throwIfInvalidWriteApiConfiguration(channelName)
+
+    // Validate input parameter
+    if (amount.isNaN() || amount.isInfinite()) {
+      throw objectError("Counter value increment should be a valid number")
+    }
+
+    // Create ObjectMessage with the COUNTER_INC operation
+    val msg = ObjectMessage(
+      operation = ObjectOperation(
+        action = ObjectOperationAction.CounterInc,
+        objectId = objectId,
+        counterOp = ObjectCounterOp(amount = amount)
+      )
+    )
+
+    // Publish the message
+    liveObjects.publish(arrayOf(msg))
+  }
+
   override fun applyObjectState(objectState: ObjectState, message: ObjectMessage): LiveCounterUpdate {
     return liveCounterManager.applyState(objectState, message.serialTimestamp)
   }

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

@@ -15,6 +15,7 @@ import io.ably.lib.objects.type.map.LiveMapValue
 import io.ably.lib.objects.type.noOp
 import io.ably.lib.types.Callback
 import io.ably.lib.util.Log
+import kotlinx.coroutines.runBlocking
 import java.util.concurrent.ConcurrentHashMap
 import java.util.AbstractMap
 
@@ -44,6 +45,7 @@ internal class DefaultLiveMap private constructor(
   private val channelName = liveObjects.channelName
   private val adapter: LiveObjectsAdapter get() = liveObjects.adapter
   internal val objectsPool: ObjectsPool get() = liveObjects.objectsPool
+  private val asyncScope get() = liveObjects.asyncScope
 
   override fun get(keyName: String): LiveMapValue? {
     adapter.throwIfInvalidAccessApiConfiguration(channelName) // RTLM5b, RTLM5c
@@ -92,20 +94,16 @@ internal class DefaultLiveMap private constructor(
     return data.values.count { !it.isEntryOrRefTombstoned(objectsPool) }.toLong() // RTLM10d
   }
 
-  override fun set(keyName: String, value: LiveMapValue) {
-    TODO("Not yet implemented")
-  }
+  override fun set(keyName: String, value: LiveMapValue) = runBlocking { setAsync(keyName, value) }
 
-  override fun remove(keyName: String) {
-    TODO("Not yet implemented")
-  }
+  override fun remove(keyName: String) = runBlocking { removeAsync(keyName) }
 
-  override fun setAsync(keyName: String, value: Any, callback: Callback<Void>) {
-    TODO("Not yet implemented")
+  override fun setAsync(keyName: String, value: LiveMapValue, callback: Callback<Void>) {
+    asyncScope.launchWithVoidCallback(callback) { setAsync(keyName, value) }
   }
 
   override fun removeAsync(keyName: String, callback: Callback<Void>) {
-    TODO("Not yet implemented")
+    asyncScope.launchWithVoidCallback(callback) { removeAsync(keyName) }
   }
 
   override fun validate(state: ObjectState) = liveMapManager.validate(state)
@@ -119,6 +117,53 @@ internal class DefaultLiveMap private constructor(
 
   override fun unsubscribeAll() = liveMapManager.unsubscribeAll()
 
+  private suspend fun setAsync(keyName: String, value: LiveMapValue) {
+    // Validate write API configuration
+    adapter.throwIfInvalidWriteApiConfiguration(channelName)
+
+    // Validate input parameters
+    if (keyName.isEmpty()) {
+      throw objectError("Map key should not be empty")
+    }
+
+    // Create ObjectMessage with the MAP_SET operation
+    val msg = ObjectMessage(
+      operation = ObjectOperation(
+        action = ObjectOperationAction.MapSet,
+        objectId = objectId,
+        mapOp = ObjectMapOp(
+          key = keyName,
+          data = fromLiveMapValue(value)
+        )
+      )
+    )
+
+    // Publish the message
+    liveObjects.publish(arrayOf(msg))
+  }
+
+  private suspend fun removeAsync(keyName: String) {
+    // Validate write API configuration
+    adapter.throwIfInvalidWriteApiConfiguration(channelName)
+
+    // Validate input parameter
+    if (keyName.isEmpty()) {
+      throw objectError("Map key should not be empty")
+    }
+
+    // Create ObjectMessage with the MAP_REMOVE operation
+    val msg = ObjectMessage(
+      operation = ObjectOperation(
+        action = ObjectOperationAction.MapRemove,
+        objectId = objectId,
+        mapOp = ObjectMapOp(key = keyName)
+      )
+    )
+
+    // Publish the message
+    liveObjects.publish(arrayOf(msg))
+  }
+
   override fun applyObjectState(objectState: ObjectState, message: ObjectMessage): LiveMapUpdate {
     return liveMapManager.applyState(objectState, message.serialTimestamp)
   }
@@ -169,5 +214,12 @@ internal class DefaultLiveMap private constructor(
         )
       )
     }
+
+    private 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/test/kotlin/io/ably/lib/objects/integration/DefaultLiveCounterTest.kt

@@ -5,6 +5,7 @@ import io.ably.lib.objects.integration.helpers.ObjectId
 import io.ably.lib.objects.integration.helpers.fixtures.createUserEngagementMatrixMap
 import io.ably.lib.objects.integration.helpers.fixtures.createUserMapWithCountersObject
 import io.ably.lib.objects.integration.setup.IntegrationTest
+import io.ably.lib.objects.type.map.LiveMapValue
 import kotlinx.coroutines.test.runTest
 import org.junit.Test
 import kotlin.test.assertEquals
@@ -212,8 +213,8 @@ class DefaultLiveCounterTest: IntegrationTest() {
     val rootMap = channel.objects.root
 
     // Step 1: Create a new counter with initial value of 10
-    val testCounterObjectId = objects.createCounter( 10.0)
-    restObjects.setMapRef(channelName, "root", "testCounter", testCounterObjectId.ObjectId)
+    val testCounterObject = objects.createCounter( 10.0)
+    rootMap.set("testCounter", LiveMapValue.of(testCounterObject))
 
     // Wait for updated testCounter to be available in the root map
     assertWaiter { rootMap.get("testCounter") != null }
@@ -222,6 +223,77 @@ class DefaultLiveCounterTest: IntegrationTest() {
     val testCounter = rootMap.get("testCounter")?.asLiveCounter
     assertNotNull(testCounter, "Test counter should be created and accessible")
     assertEquals(10.0, testCounter.value(), "Counter should have initial value of 10")
+
+    // Step 2: Increment counter by 5 (10 + 5 = 15)
+    testCounter.increment(5.0)
+    // Wait for the counter to be updated
+    assertWaiter { testCounter.value() == 15.0 }
+
+    // Assert after first increment
+    assertEquals(15.0, testCounter.value(), "Counter should be incremented to 15")
+
+    // Step 3: Increment counter by 3 (15 + 3 = 18)
+    testCounter.increment(3.0)
+    // Wait for the counter to be updated
+    assertWaiter { testCounter.value() == 18.0 }
+
+    // Assert after second increment
+    assertEquals(18.0, testCounter.value(), "Counter should be incremented to 18")
+
+    // Step 4: Increment counter by a larger amount: 12 (18 + 12 = 30)
+    testCounter.increment(12.0)
+    // Wait for the counter to be updated
+    assertWaiter { testCounter.value() == 30.0 }
+
+    // Assert after third increment
+    assertEquals(30.0, testCounter.value(), "Counter should be incremented to 30")
+
+    // Step 5: Decrement counter by 7 (30 - 7 = 23)
+    testCounter.decrement(7.0)
+    // Wait for the counter to be updated
+    assertWaiter { testCounter.value() == 23.0 }
+
+    // Assert after first decrement
+    assertEquals(23.0, testCounter.value(), "Counter should be decremented to 23")
+
+    // Step 6: Decrement counter by 4 (23 - 4 = 19)
+    testCounter.decrement(4.0)
+    // Wait for the counter to be updated
+    assertWaiter { testCounter.value() == 19.0 }
+
+    // Assert after second decrement
+    assertEquals(19.0, testCounter.value(), "Counter should be decremented to 19")
+
+    // Step 7: Increment counter by 1 (19 + 1 = 20)
+    testCounter.increment(1.0)
+    // Wait for the counter to be updated
+    assertWaiter { testCounter.value() == 20.0 }
+
+    // Assert after final increment
+    assertEquals(20.0, testCounter.value(), "Counter should be incremented to 20")
+
+    // Step 8: Decrement counter by a larger amount: 15 (20 - 15 = 5)
+    testCounter.decrement(15.0)
+    // Wait for the counter to be updated
+    assertWaiter { testCounter.value() == 5.0 }
+
+    // Assert after large decrement
+    assertEquals(5.0, testCounter.value(), "Counter should be decremented to 5")
+
+    // Final verification - test final increment to ensure counter still works
+    testCounter.increment(25.0)
+    assertWaiter { testCounter.value() == 30.0 }
+
+    // Assert final state
+    assertEquals(30.0, testCounter.value(), "Counter should have final value of 30")
+
+    // Verify the counter object is still accessible and functioning
+    assertNotNull(testCounter, "Counter should still be accessible at the end")
+
+    // Verify we can still access it from the root map
+    val finalCounterCheck = rootMap.get("testCounter")?.asLiveCounter
+    assertNotNull(finalCounterCheck, "Counter should still be accessible from root map")
+    assertEquals(30.0, finalCounterCheck.value(), "Final counter value should be 30 when accessed from root map")
   }
 
   @Test