[ECO-5458] Added comprehensive integration tests for subscriptions

- Enhanced DefaultLiveCounterTest with subscription test scenarios
- Updated DefaultLiveMapTest with subscription functionality tests
- Added subscription tests to DefaultLiveObjectsTest and helper utilities

Author: sacOO7

live-objects/src/test/kotlin/io/ably/lib/objects/integration/DefaultLiveCounterTest.kt

@@ -1,14 +1,17 @@
 package io.ably.lib.objects.integration
 
-import io.ably.lib.objects.LiveCounter
-import io.ably.lib.objects.LiveMap
+import io.ably.lib.objects.type.counter.LiveCounter
+import io.ably.lib.objects.type.map.LiveMap
 import io.ably.lib.objects.assertWaiter
+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 kotlinx.coroutines.test.runTest
 import org.junit.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
 
 class DefaultLiveCounterTest: IntegrationTest() {
   /**
@@ -202,4 +205,73 @@ class DefaultLiveCounterTest: IntegrationTest() {
     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
+  fun testLiveCounterChangesUsingSubscription() = runTest {
+    val channelName = generateChannelName()
+    val userEngagementMapId = restObjects.createUserEngagementMatrixMap(channelName)
+    restObjects.setMapRef(channelName, "root", "userMatrix", userEngagementMapId)
+
+    val channel = getRealtimeChannel(channelName)
+    val rootMap = channel.objects.root
+
+    val userEngagementMap = rootMap.get("userMatrix") as LiveMap
+    assertEquals(4L, userEngagementMap.size(), "User engagement map should contain 4 top-level entries")
+
+    val totalReactions = userEngagementMap.get("totalReactions") as LiveCounter
+    assertEquals(189.0, totalReactions.value(), "Total reactions counter should have initial value of 189")
+
+    // Subscribe to changes on the totalReactions counter
+    val counterUpdates = mutableListOf<Double>()
+    val totalReactionsSubscription = totalReactions.subscribe { update ->
+      counterUpdates.add(update.update.amount)
+    }
+
+    // Step 1: Increment the totalReactions counter by 10 (189 + 10 = 199)
+    restObjects.incrementCounter(channelName, totalReactions.ObjectId, 10.0)
+
+    // Wait for the update to be received
+    assertWaiter { counterUpdates.isNotEmpty() }
+
+    // Verify the increment update was received
+    assertEquals(1, counterUpdates.size, "Should receive one update for increment")
+    assertEquals(10.0, counterUpdates.first(), "Update should contain increment amount of 10")
+    assertEquals(199.0, totalReactions.value(), "Counter should be incremented to 199")
+
+    // Step 2: Decrement the totalReactions counter by 5 (199 - 5 = 194)
+    counterUpdates.clear()
+    restObjects.decrementCounter(channelName, totalReactions.ObjectId, 5.0)
+
+    // Wait for the second update
+    assertWaiter { counterUpdates.isNotEmpty() }
+
+    // Verify the decrement update was received
+    assertEquals(1, counterUpdates.size, "Should receive one update for decrement")
+    assertEquals(-5.0, counterUpdates.first(), "Update should contain decrement amount of -5")
+    assertEquals(194.0, totalReactions.value(), "Counter should be decremented to 194")
+
+    // Step 3: Increment the totalReactions counter by 15 (194 + 15 = 209)
+    counterUpdates.clear()
+    restObjects.incrementCounter(channelName, totalReactions.ObjectId, 15.0)
+
+    // Wait for the third update
+    assertWaiter { counterUpdates.isNotEmpty() }
+
+    // Verify the third increment update was received
+    assertEquals(1, counterUpdates.size, "Should receive one update for third increment")
+    assertEquals(15.0, counterUpdates.first(), "Update should contain increment amount of 15")
+    assertEquals(209.0, totalReactions.value(), "Counter should be incremented to 209")
+
+    // Clean up subscription
+    counterUpdates.clear()
+    totalReactionsSubscription.unsubscribe()
+
+    // No updates should be received after unsubscribing
+    restObjects.incrementCounter(channelName, totalReactions.ObjectId, 20.0)
+
+    // Wait for a moment to ensure no updates are received
+    assertWaiter { totalReactions.value() == 229.0 }
+
+    assertTrue(counterUpdates.isEmpty(), "No updates should be received after unsubscribing")
+  }
 }

live-objects/src/test/kotlin/io/ably/lib/objects/integration/DefaultLiveMapTest.kt

@@ -4,12 +4,17 @@ import io.ably.lib.objects.*
 import io.ably.lib.objects.ObjectData
 import io.ably.lib.objects.ObjectValue
 import io.ably.lib.objects.integration.helpers.fixtures.createUserMapObject
+import io.ably.lib.objects.integration.helpers.fixtures.createUserProfileMapObject
 import io.ably.lib.objects.integration.setup.IntegrationTest
+import io.ably.lib.objects.type.counter.LiveCounter
+import io.ably.lib.objects.type.map.LiveMap
+import io.ably.lib.objects.type.map.LiveMapUpdate
 import kotlinx.coroutines.test.runTest
 import org.junit.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertNotNull
 import kotlin.test.assertNull
+import kotlin.test.assertTrue
 
 class DefaultLiveMapTest: IntegrationTest() {
   /**
@@ -210,4 +215,94 @@ class DefaultLiveMapTest: IntegrationTest() {
     val finalValues = testMap.values().toSet()
     assertEquals(setOf("Bob", false, "bob@example.com"), finalValues, "Final values should match expected set")
   }
+
+  @Test
+  fun testLiveMapChangesUsingSubscription() = runTest {
+    val channelName = generateChannelName()
+    val userProfileObjectId = restObjects.createUserProfileMapObject(channelName)
+    restObjects.setMapRef(channelName, "root", "userProfile", userProfileObjectId)
+
+    val channel = getRealtimeChannel(channelName)
+    val rootMap = channel.objects.root
+
+    // Get the user profile map object from the root map
+    val userProfile = rootMap.get("userProfile") as LiveMap
+    assertNotNull(userProfile, "User profile should be synchronized")
+    assertEquals(4L, userProfile.size(), "User profile should contain 4 entries")
+
+    // Verify initial values
+    assertEquals("user123", userProfile.get("userId"), "Initial userId should be user123")
+    assertEquals("John Doe", userProfile.get("name"), "Initial name should be John Doe")
+    assertEquals("john@example.com", userProfile.get("email"), "Initial email should be john@example.com")
+    assertEquals(true, userProfile.get("isActive"), "Initial isActive should be true")
+
+    // Subscribe to changes in the user profile map
+    val userProfileUpdates = mutableListOf<LiveMapUpdate>()
+    val userProfileSubscription = userProfile.subscribe { update -> userProfileUpdates.add(update) }
+
+    // Step 1: Update an existing field in the user profile map (change the name)
+    restObjects.setMapValue(channelName, userProfileObjectId, "name", ObjectValue.String("Bob Smith"))
+
+    // Wait for the update to be received
+    assertWaiter { userProfileUpdates.isNotEmpty() }
+
+    // Verify the update was received
+    assertEquals(1, userProfileUpdates.size, "Should receive one update")
+    val firstUpdateMap = userProfileUpdates.first().update
+    assertEquals(1, firstUpdateMap.size, "Should have one key change")
+    assertTrue(firstUpdateMap.containsKey("name"), "Update should contain name key")
+    assertEquals(LiveMapUpdate.Change.UPDATED, firstUpdateMap["name"], "name should be marked as UPDATED")
+
+    // Verify the value was actually updated
+    assertEquals("Bob Smith", userProfile.get("name"), "Name should be updated to Bob Smith")
+
+    // Step 2: Update another field in the user profile map (change the email)
+    userProfileUpdates.clear()
+    restObjects.setMapValue(channelName, userProfileObjectId, "email", ObjectValue.String("bob@example.com"))
+
+    // Wait for the second update
+    assertWaiter { userProfileUpdates.isNotEmpty() }
+
+    // Verify the second update
+    assertEquals(1, userProfileUpdates.size, "Should receive one update for the second change")
+    val secondUpdateMap = userProfileUpdates.first().update
+    assertEquals(1, secondUpdateMap.size, "Should have one key change")
+    assertTrue(secondUpdateMap.containsKey("email"), "Update should contain email key")
+    assertEquals(LiveMapUpdate.Change.UPDATED, secondUpdateMap["email"], "email should be marked as UPDATED")
+
+    // Verify the value was actually updated
+    assertEquals("bob@example.com", userProfile.get("email"), "Email should be updated to bob@example.com")
+
+    // Step 3: Remove an existing field from the user profile map (remove isActive)
+    userProfileUpdates.clear()
+    restObjects.removeMapValue(channelName, userProfileObjectId, "isActive")
+
+    // Wait for the removal update
+    assertWaiter { userProfileUpdates.isNotEmpty() }
+
+    // Verify the removal update
+    assertEquals(1, userProfileUpdates.size, "Should receive one update for removal")
+    val removalUpdateMap = userProfileUpdates.first().update
+    assertEquals(1, removalUpdateMap.size, "Should have one key change")
+    assertTrue(removalUpdateMap.containsKey("isActive"), "Update should contain isActive key")
+    assertEquals(LiveMapUpdate.Change.REMOVED, removalUpdateMap["isActive"], "isActive should be marked as REMOVED")
+
+    // Verify final state of the user profile map
+    assertEquals(3L, userProfile.size(), "User profile should have 3 entries after removing isActive")
+    assertEquals("user123", userProfile.get("userId"), "userId should remain unchanged")
+    assertEquals("Bob Smith", userProfile.get("name"), "name should remain updated")
+    assertEquals("bob@example.com", userProfile.get("email"), "email should remain updated")
+    assertNull(userProfile.get("isActive"), "isActive should be removed")
+
+    // Clean up subscription
+    userProfileUpdates.clear()
+    userProfileSubscription.unsubscribe()
+    // No updates should be received after unsubscribing
+    restObjects.setMapValue(channelName, userProfileObjectId, "country", ObjectValue.String("uk"))
+
+    // Wait for a moment to ensure no updates are received
+    assertWaiter { userProfile.size() == 4L }
+
+    assertTrue(userProfileUpdates.isEmpty(), "No updates should be received after unsubscribing")
+  }
 }

live-objects/src/test/kotlin/io/ably/lib/objects/integration/DefaultLiveObjectsTest.kt

@@ -9,6 +9,8 @@ import io.ably.lib.objects.integration.helpers.fixtures.initializeRootMap
 import io.ably.lib.objects.integration.setup.IntegrationTest
 import io.ably.lib.objects.size
 import io.ably.lib.objects.state.ObjectsStateEvent
+import io.ably.lib.objects.type.counter.LiveCounter
+import io.ably.lib.objects.type.map.LiveMap
 import kotlinx.coroutines.test.runTest
 import org.junit.Test
 import kotlin.test.assertEquals

live-objects/src/test/kotlin/io/ably/lib/objects/integration/helpers/RestObjects.kt

@@ -5,6 +5,7 @@ import io.ably.lib.objects.ObjectData
 import io.ably.lib.objects.ObjectValue
 import io.ably.lib.rest.AblyRest
 import io.ably.lib.http.HttpUtils
+import io.ably.lib.objects.integration.helpers.fixtures.DataFixtures
 import io.ably.lib.types.ClientOptions
 
 /**
@@ -37,8 +38,7 @@ internal class RestObjects(options: ClientOptions) {
    * Sets an object reference at the specified key in an existing map.
    */
   internal fun setMapRef(channelName: String, mapObjectId: String, key: String, refMapObjectId: String) {
-    val data = ObjectData(objectId = refMapObjectId)
-    val mapCreateOp = PayloadBuilder.mapSetRestOp(mapObjectId, key, data)
+    val mapCreateOp = PayloadBuilder.mapSetRestOp(mapObjectId, key, DataFixtures.mapRef(refMapObjectId))
     operationRequest(channelName, mapCreateOp)
   }
 

live-objects/src/test/kotlin/io/ably/lib/objects/integration/helpers/Utils.kt

@@ -1,8 +1,8 @@
 package io.ably.lib.objects.integration.helpers
 
 import io.ably.lib.objects.DefaultLiveObjects
-import io.ably.lib.objects.LiveCounter
-import io.ably.lib.objects.LiveMap
+import io.ably.lib.objects.type.counter.LiveCounter
+import io.ably.lib.objects.type.map.LiveMap
 import io.ably.lib.objects.LiveObjects
 import io.ably.lib.objects.type.livecounter.DefaultLiveCounter
 import io.ably.lib.objects.type.livemap.DefaultLiveMap