re-port

Author: lawrence-forooghian

Tests/AblyLiveObjectsTests/JS Integration Tests/ObjectsIntegrationTests.swift

@@ -3467,51 +3467,40 @@ private struct ObjectsIntegrationTests {
                     let counter = try #require(ctx.root.get(key: ctx.sampleCounterKey)?.liveCounterValue)
                     let callbacks = 3
                     let callbacksCalled = MainActorStorage<[Int]>(value: Array(repeating: 0, count: callbacks))
-                    
-                    // Create multiple subscribers
-                    let subscriber0 = Subscriber<LiveCounterUpdate, SubscribeResponse>(callbackQueue: .main)
-                    let subscriber1 = Subscriber<LiveCounterUpdate, SubscribeResponse>(callbackQueue: .main)
-                    let subscriber2 = Subscriber<LiveCounterUpdate, SubscribeResponse>(callbackQueue: .main)
-                    
-                    try counter.subscribe(listener: subscriber0.createListener())
-                    try counter.subscribe(listener: subscriber1.createListener())
-                    try counter.subscribe(listener: subscriber2.createListener())
-                    
-                    // Set up async let promises for each subscriber
-                    async let subscriptionPromise0: Void = withCheckedContinuation { continuation in
-                        subscriber0.addListener { _, _ in
-                            MainActor.assumeIsolated {
-                                callbacksCalled.value[0] += 1
-                                continuation.resume()
-                            }
-                        }
-                    }
-                    
-                    async let subscriptionPromise1: Void = withCheckedContinuation { continuation in
-                        subscriber1.addListener { _, _ in
-                            MainActor.assumeIsolated {
-                                callbacksCalled.value[1] += 1
-                                continuation.resume()
-                            }
-                        }
+
+                    // Create multiple subscribers synchronously
+                    let subscribers = try (0 ..< callbacks).map { _ in
+                        let subscriber = Subscriber<LiveCounterUpdate, SubscribeResponse>(callbackQueue: .main)
+                        try counter.subscribe(listener: subscriber.createListener())
+                        return subscriber
                     }
-                    
-                    async let subscriptionPromise2: Void = withCheckedContinuation { continuation in
-                        subscriber2.addListener { _, _ in
-                            MainActor.assumeIsolated {
-                                callbacksCalled.value[2] += 1
-                                continuation.resume()
+
+                    // Set up subscription promises using TaskGroup
+                    async let subscriptionPromises: Void = withTaskGroup(of: Void.self) { group in
+                        for (index, subscriber) in subscribers.enumerated() {
+                            group.addTask {
+                                await withCheckedContinuation { continuation in
+                                    subscriber.addListener { _, _ in
+                                        MainActor.assumeIsolated {
+                                            callbacksCalled.value[index] += 1
+                                            continuation.resume()
+                                        }
+                                    }
+                                }
                             }
                         }
+
+                        // Wait for all subscription tasks to complete
+                        for await _ in group {}
                     }
 
                     let increments = 3
-                    for i in 0..<increments {
+                    for i in 0 ..< increments {
                         let counterUpdatesStream = try counter.updates()
                         async let counterUpdatedPromise: Void = waitForCounterUpdate(counterUpdatesStream)
                         _ = try await ctx.objectsHelper.operationRequest(
                             channelName: ctx.channelName,
-                            opBody: ctx.objectsHelper.counterIncRestOp(objectId: ctx.sampleCounterObjectId, number: 1)
+                            opBody: ctx.objectsHelper.counterIncRestOp(objectId: ctx.sampleCounterObjectId, number: 1),
                         )
                         await counterUpdatedPromise
 
@@ -3522,12 +3511,10 @@ private struct ObjectsIntegrationTests {
                     }
 
                     // Wait for all subscription promises to complete
-                    await subscriptionPromise0
-                    await subscriptionPromise1
-                    await subscriptionPromise2
+                    await subscriptionPromises
 
                     #expect(try counter.value == 3, "Check counter has final expected value after all increments")
-                    for i in 0..<callbacks {
+                    for i in 0 ..< callbacks {
                         #expect(callbacksCalled.value[i] == 1, "Check subscription callback \(i) was called once")
                     }
                 },
@@ -3586,55 +3573,44 @@ private struct ObjectsIntegrationTests {
                     let map = try #require(ctx.root.get(key: ctx.sampleMapKey)?.liveMapValue)
                     let callbacks = 3
                     let callbacksCalled = MainActorStorage<[Int]>(value: Array(repeating: 0, count: callbacks))
-                    
-                    // Create multiple subscribers
-                    let subscriber0 = Subscriber<LiveMapUpdate, SubscribeResponse>(callbackQueue: .main)
-                    let subscriber1 = Subscriber<LiveMapUpdate, SubscribeResponse>(callbackQueue: .main)
-                    let subscriber2 = Subscriber<LiveMapUpdate, SubscribeResponse>(callbackQueue: .main)
-                    
-                    try map.subscribe(listener: subscriber0.createListener())
-                    try map.subscribe(listener: subscriber1.createListener())
-                    try map.subscribe(listener: subscriber2.createListener())
-                    
-                    // Set up async let promises for each subscriber
-                    async let subscriptionPromise0: Void = withCheckedContinuation { continuation in
-                        subscriber0.addListener { _, _ in
-                            MainActor.assumeIsolated {
-                                callbacksCalled.value[0] += 1
-                                continuation.resume()
-                            }
-                        }
-                    }
-                    
-                    async let subscriptionPromise1: Void = withCheckedContinuation { continuation in
-                        subscriber1.addListener { _, _ in
-                            MainActor.assumeIsolated {
-                                callbacksCalled.value[1] += 1
-                                continuation.resume()
-                            }
-                        }
+
+                    // Create multiple subscribers synchronously
+                    let subscribers = try (0 ..< callbacks).map { _ in
+                        let subscriber = Subscriber<LiveMapUpdate, SubscribeResponse>(callbackQueue: .main)
+                        try map.subscribe(listener: subscriber.createListener())
+                        return subscriber
                     }
-                    
-                    async let subscriptionPromise2: Void = withCheckedContinuation { continuation in
-                        subscriber2.addListener { _, _ in
-                            MainActor.assumeIsolated {
-                                callbacksCalled.value[2] += 1
-                                continuation.resume()
+
+                    // Set up subscription promises using TaskGroup
+                    async let subscriptionPromises: Void = withTaskGroup(of: Void.self) { group in
+                        for (index, subscriber) in subscribers.enumerated() {
+                            group.addTask {
+                                await withCheckedContinuation { continuation in
+                                    subscriber.addListener { _, _ in
+                                        MainActor.assumeIsolated {
+                                            callbacksCalled.value[index] += 1
+                                            continuation.resume()
+                                        }
+                                    }
+                                }
                             }
                         }
+
+                        // Wait for all subscription tasks to complete
+                        for await _ in group {}
                     }
 
                     let mapSets = 3
-                    for i in 0..<mapSets {
+                    for i in 0 ..< mapSets {
                         let mapUpdatesStream = try map.updates()
                         async let mapUpdatedPromise: Void = waitForMapKeyUpdate(mapUpdatesStream, "foo-\(i)")
                         _ = try await ctx.objectsHelper.operationRequest(
                             channelName: ctx.channelName,
                             opBody: ctx.objectsHelper.mapSetRestOp(
                                 objectId: ctx.sampleMapObjectId,
                                 key: "foo-\(i)",
-                                value: ["string": "exists"]
-                            )
+                                value: ["string": "exists"],
+                            ),
                         )
                         await mapUpdatedPromise
 
@@ -3645,15 +3621,13 @@ private struct ObjectsIntegrationTests {
                     }
 
                     // Wait for all subscription promises to complete
-                    await subscriptionPromise0
-                    await subscriptionPromise1
-                    await subscriptionPromise2
+                    await subscriptionPromises
 
-                    for i in 0..<mapSets {
+                    for i in 0 ..< mapSets {
                         let value = try #require(map.get(key: "foo-\(i)")?.stringValue)
                         #expect(value == "exists", "Check map has value for key \"foo-\(i)\" after all map sets")
                     }
-                    for i in 0..<callbacks {
+                    for i in 0 ..< callbacks {
                         #expect(callbacksCalled.value[i] == 1, "Check subscription callback \(i) was called once")
                     }
                 },

prompt-re-port-two-subscription-scenarios.md

@@ -26,7 +26,7 @@ Begin by creating a TODO list for the remaining tests, then work on this TODO li
 Porting a couple of patterns used in the JS tests (see existing examples in ObjectsIntegrationTests.swift):
 
 - a subscription listener that manipulates a local variable stored outside the listener: use a local variable of type MainActorStorage, combined with `MainActor.assumeIsolated` inside the listener
-- an `async let` task that waits for a subscription listener to be invoked: create a `Subscriber(callbackQueue: .main)` synchronously (so that no events are missed) and then call `.addListener(:_)` on this `Subscription` inside the `withCheckedContinuation` closure 
+- an `async let` task that waits for a subscription listener to be invoked: create a `Subscriber(callbackQueue: .main)` synchronously (so that no events are missed) and then call `.addListener(:_)` on this `Subscription` inside the `withCheckedContinuation` closure
 
 ## Detailed general requirements for the ported tests