Make integration test subscription behaviour consistent with JS tests

Make sure that we perform the subscription synchronously so as not to
miss events. This is a nuance that I missed when looking at Cursor's
generated code in fa255c1.

The way in which I've chosen to do this is not particularly elegant (and
to be honest Swift doesn't make it very easy to do so) but I think that
it's the closest to the JS code, which hopefully will mean Cursor will
be able to easily apply this pattern when translating further JS tests.

Author: lawrence-forooghian

Tests/AblyLiveObjectsTests/JS Integration Tests/ObjectsIntegrationTests.swift

@@ -88,32 +88,12 @@ func waitFixtureChannelIsReady(_: ARTRealtime) async throws {
     try await Task.sleep(nanoseconds: 5 * NSEC_PER_SEC)
 }
 
-func waitForMapKeyUpdate(_ map: any LiveMap, _ key: String) async throws {
-    try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, _>) in
-        do {
-            try map.subscribe { update, subscription in
-                if update.update[key] != nil {
-                    subscription.unsubscribe()
-                    continuation.resume()
-                }
-            }
-        } catch {
-            continuation.resume(throwing: error)
-        }
-    }
+func waitForMapKeyUpdate(_ updates: AsyncStream<LiveMapUpdate>, _ key: String) async {
+    _ = await updates.first { $0.update[key] != nil }
 }
 
-func waitForCounterUpdate(_ counter: any LiveCounter) async throws {
-    try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, _>) in
-        do {
-            try counter.subscribe { _, subscription in
-                subscription.unsubscribe()
-                continuation.resume()
-            }
-        } catch {
-            continuation.resume(throwing: error)
-        }
-    }
+func waitForCounterUpdate(_ updates: AsyncStream<LiveCounterUpdate>) async {
+    _ = await updates.first { _ in true }
 }
 
 // I added this @MainActor as an "I don't understand what's going on there; let's try this" when observing that for some reason the setter of setListenerAfterProcessingIncomingMessage was hanging inside `-[ARTSRDelegateController dispatchQueue]`. This seems to avoid it and I have not investigated more deeply 🤷
@@ -321,12 +301,14 @@ private struct ObjectsIntegrationTests {
                         let objects = ctx.objects
 
                         // Create the promise first, before the operations that will trigger it
+                        let objectsCreatedPromiseUpdates1 = try root.updates()
+                        let objectsCreatedPromiseUpdates2 = try root.updates()
                         async let objectsCreatedPromise: Void = withThrowingTaskGroup(of: Void.self) { group in
                             group.addTask {
-                                try await waitForMapKeyUpdate(root, "counter")
+                                await waitForMapKeyUpdate(objectsCreatedPromiseUpdates1, "counter")
                             }
                             group.addTask {
-                                try await waitForMapKeyUpdate(root, "map")
+                                await waitForMapKeyUpdate(objectsCreatedPromiseUpdates2, "map")
                             }
                             while try await group.next() != nil {}
                         }
@@ -342,15 +324,18 @@ private struct ObjectsIntegrationTests {
                         _ = try await (setMapPromise, setCounterPromise, objectsCreatedPromise)
 
                         // Create the promise first, before the operations that will trigger it
+                        let operationsAppliedPromiseUpdates1 = try map.updates()
+                        let operationsAppliedPromiseUpdates2 = try map.updates()
+                        let operationsAppliedPromiseUpdates3 = try counter.updates()
                         async let operationsAppliedPromise: Void = withThrowingTaskGroup(of: Void.self) { group in
                             group.addTask {
-                                try await waitForMapKeyUpdate(map, "anotherKey")
+                                await waitForMapKeyUpdate(operationsAppliedPromiseUpdates1, "anotherKey")
                             }
                             group.addTask {
-                                try await waitForMapKeyUpdate(map, "shouldDelete")
+                                await waitForMapKeyUpdate(operationsAppliedPromiseUpdates2, "shouldDelete")
                             }
                             group.addTask {
-                                try await waitForCounterUpdate(counter)
+                                await waitForCounterUpdate(operationsAppliedPromiseUpdates3)
                             }
                             while try await group.next() != nil {}
                         }
@@ -393,12 +378,14 @@ private struct ObjectsIntegrationTests {
                         let client = ctx.client
 
                         // Create the promise first, before the operations that will trigger it
+                        let objectsCreatedPromiseUpdates1 = try root.updates()
+                        let objectsCreatedPromiseUpdates2 = try root.updates()
                         async let objectsCreatedPromise: Void = withThrowingTaskGroup(of: Void.self) { group in
                             group.addTask {
-                                try await waitForMapKeyUpdate(root, "counter")
+                                await waitForMapKeyUpdate(objectsCreatedPromiseUpdates1, "counter")
                             }
                             group.addTask {
-                                try await waitForMapKeyUpdate(root, "map")
+                                await waitForMapKeyUpdate(objectsCreatedPromiseUpdates2, "map")
                             }
                             while try await group.next() != nil {}
                         }
@@ -588,14 +575,15 @@ private struct ObjectsIntegrationTests {
                         let channelName = ctx.channelName
                         let channel = ctx.channel
 
-                        async let counterCreatedPromise: Void = waitForMapKeyUpdate(root, "counter")
+                        let counterCreatedPromiseUpdates = try root.updates()
+                        async let counterCreatedPromise: Void = waitForMapKeyUpdate(counterCreatedPromiseUpdates, "counter")
                         let counterResult = try await objectsHelper.createAndSetOnMap(
                             channelName: channelName,
                             mapObjectId: "root",
                             key: "counter",
                             createOp: objectsHelper.counterCreateRestOp(number: 1),
                         )
-                        _ = try await counterCreatedPromise
+                        _ = await counterCreatedPromise
 
                         #expect(try root.get(key: "counter") != nil, "Check counter exists on root before OBJECT_SYNC sequence with \"tombstone=true\"")
 
@@ -642,25 +630,21 @@ private struct ObjectsIntegrationTests {
                         let channelName = ctx.channelName
                         let channel = ctx.channel
 
-                        async let counterCreatedPromise: Void = waitForMapKeyUpdate(root, "counter")
+                        let counterCreatedPromiseUpdates = try root.updates()
+                        async let counterCreatedPromise: Void = waitForMapKeyUpdate(counterCreatedPromiseUpdates, "counter")
                         let counterResult = try await objectsHelper.createAndSetOnMap(
                             channelName: channelName,
                             mapObjectId: "root",
                             key: "counter",
                             createOp: objectsHelper.counterCreateRestOp(number: 1),
                         )
-                        _ = try await counterCreatedPromise
-
-                        async let counterSubPromise: Void = withCheckedThrowingContinuation { continuation in
-                            do {
-                                try #require(root.get(key: "counter")?.liveCounterValue).subscribe { update, _ in
-                                    #expect(update.amount == -1, "Check counter subscription callback is called with an expected update object after OBJECT_SYNC sequence with \"tombstone=true\"")
-                                    continuation.resume()
-                                }
-                            } catch {
-                                continuation.resume(throwing: error)
-                            }
-                        }
+                        _ = await counterCreatedPromise
+
+                        let counterSubPromiseUpdates = try #require(root.get(key: "counter")?.liveCounterValue).updates()
+                        async let counterSubPromise: Void = {
+                            let update = try await #require(counterSubPromiseUpdates.first { _ in true })
+                            #expect(update.amount == -1, "Check counter subscription callback is called with an expected update object after OBJECT_SYNC sequence with \"tombstone=true\"")
+                        }()
 
                         // inject an OBJECT_SYNC message where a counter is now tombstoned
                         try await objectsHelper.processObjectStateMessageOnChannel(