Make callbacks Sendable

Didn't do this in ce8c022 because I didn't have a good idea of our
threading approach. But for the initial approach that we'll be taking —
namely, calling the callbacks on ably-cocoa's callback queue — I think
we'll need it.

Haven't done it for batch stuff yet because I don't yet know whether
it'll be necessary (will get a better idea when we implement it).

Now that the callback can be called on any thread, we can no longer
easily use the "capture the return value of `subscribe` pattern so that
we can unsubscribe later" pattern. So, I've decided to pass the
subscripiton as a second argument to the callback. Might be there's a
better pattern we can use (e.g. pass in an object to the `subscribe`
call, like JS `fetch`'s AbortController), but this'll do for now.

Author: lawrence-forooghian

Sources/AblyLiveObjects/Public/Public Proxy Objects/PublicDefaultLiveCounter.swift

@@ -34,15 +34,15 @@ internal final class PublicDefaultLiveCounter: LiveCounter {
         try await proxied.decrement(amount: amount)
     }
 
-    internal func subscribe(listener: sending @escaping LiveObjectUpdateCallback<LiveCounterUpdate>) -> any SubscribeResponse {
+    internal func subscribe(listener: @escaping LiveObjectUpdateCallback<LiveCounterUpdate>) -> any SubscribeResponse {
         proxied.subscribe(listener: listener)
     }
 
     internal func unsubscribeAll() {
         proxied.unsubscribeAll()
     }
 
-    internal func on(event: LiveObjectLifecycleEvent, callback: sending @escaping LiveObjectLifecycleEventCallback) -> any OnLiveObjectLifecycleEventResponse {
+    internal func on(event: LiveObjectLifecycleEvent, callback: @escaping LiveObjectLifecycleEventCallback) -> any OnLiveObjectLifecycleEventResponse {
         proxied.on(event: event, callback: callback)
     }
 

Sources/AblyLiveObjects/Public/Public Proxy Objects/PublicDefaultLiveMap.swift

@@ -80,15 +80,15 @@ internal final class PublicDefaultLiveMap: LiveMap {
         try await proxied.remove(key: key)
     }
 
-    internal func subscribe(listener: sending @escaping LiveObjectUpdateCallback<LiveMapUpdate>) -> any SubscribeResponse {
+    internal func subscribe(listener: @escaping LiveObjectUpdateCallback<LiveMapUpdate>) -> any SubscribeResponse {
         proxied.subscribe(listener: listener)
     }
 
     internal func unsubscribeAll() {
         proxied.unsubscribeAll()
     }
 
-    internal func on(event: LiveObjectLifecycleEvent, callback: sending @escaping LiveObjectLifecycleEventCallback) -> any OnLiveObjectLifecycleEventResponse {
+    internal func on(event: LiveObjectLifecycleEvent, callback: @escaping LiveObjectLifecycleEventCallback) -> any OnLiveObjectLifecycleEventResponse {
         proxied.on(event: event, callback: callback)
     }
 

Sources/AblyLiveObjects/Public/Public Proxy Objects/PublicDefaultRealtimeObjects.swift

@@ -51,7 +51,7 @@ internal final class PublicDefaultRealtimeObjects: RealtimeObjects {
         try await proxied.batch(callback: callback)
     }
 
-    internal func on(event: ObjectsEvent, callback: sending @escaping ObjectsEventCallback) -> any OnObjectsEventResponse {
+    internal func on(event: ObjectsEvent, callback: @escaping ObjectsEventCallback) -> any OnObjectsEventResponse {
         proxied.on(event: event, callback: callback)
     }
 

Sources/AblyLiveObjects/Public/PublicTypes.swift

@@ -2,14 +2,19 @@ import Ably
 
 /// A callback used in ``LiveObject`` to listen for updates to the object.
 ///
-/// - Parameter update: The update object describing the changes made to the object.
-public typealias LiveObjectUpdateCallback<T> = (_ update: sending T) -> Void
+/// - Parameters:
+///   - update: The update object describing the changes made to the object.
+///   - subscription: A ``SubscribeResponse`` object that allows the provided listener to deregister itself from future updates.
+public typealias LiveObjectUpdateCallback<T> = @Sendable (_ update: sending T, _ subscription: SubscribeResponse) -> Void
 
 /// The callback used for the events emitted by ``RealtimeObjects``.
-public typealias ObjectsEventCallback = () -> Void
+///
+/// - Parameter subscription: An ``OnObjectsEventResponse`` object that allows the provided listener to deregister itself from future updates.
+public typealias ObjectsEventCallback = @Sendable (_ subscription: OnObjectsEventResponse) -> Void
 
 /// The callback used for the lifecycle events emitted by ``LiveObject``.
-public typealias LiveObjectLifecycleEventCallback = () -> Void
+/// - Parameter subscription: A ``OnLiveObjectLifecycleEventResponse`` object that allows the provided listener to deregister itself from future updates.
+public typealias LiveObjectLifecycleEventCallback = @Sendable (_ subscription: OnLiveObjectLifecycleEventResponse) -> Void
 
 /// A function passed to ``RealtimeObjects/batch(callback:)`` to group multiple Objects operations into a single channel message.
 ///
@@ -70,7 +75,7 @@ public protocol RealtimeObjects: Sendable {
     ///   - callback: The event listener.
     /// - Returns: An ``OnObjectsEventResponse`` object that allows the provided listener to be deregistered from future updates.
     @discardableResult
-    func on(event: ObjectsEvent, callback: sending @escaping ObjectsEventCallback) -> OnObjectsEventResponse
+    func on(event: ObjectsEvent, callback: @escaping ObjectsEventCallback) -> OnObjectsEventResponse
 
     /// Deregisters all registrations, for all events and listeners.
     func offAll()
@@ -337,7 +342,7 @@ public protocol LiveObject: AnyObject, Sendable {
     /// - Parameter listener: An event listener function that is called with an update object whenever this LiveObject is updated.
     /// - Returns: A ``SubscribeResponse`` object that allows the provided listener to be deregistered from future updates.
     @discardableResult
-    func subscribe(listener: sending @escaping LiveObjectUpdateCallback<Update>) -> SubscribeResponse
+    func subscribe(listener: @escaping LiveObjectUpdateCallback<Update>) -> SubscribeResponse
 
     /// Deregisters all listeners from updates for this LiveObject.
     func unsubscribeAll()
@@ -349,7 +354,7 @@ public protocol LiveObject: AnyObject, Sendable {
     ///   - callback: The event listener.
     /// - Returns: A ``OnLiveObjectLifecycleEventResponse`` object that allows the provided listener to be deregistered from future updates.
     @discardableResult
-    func on(event: LiveObjectLifecycleEvent, callback: sending @escaping LiveObjectLifecycleEventCallback) -> OnLiveObjectLifecycleEventResponse
+    func on(event: LiveObjectLifecycleEvent, callback: @escaping LiveObjectLifecycleEventCallback) -> OnLiveObjectLifecycleEventResponse
 
     /// Deregisters all registrations, for all events and listeners.
     func offAll()

Tests/AblyLiveObjectsTests/JS Integration Tests/ObjectsIntegrationTests.swift

@@ -90,8 +90,7 @@ func waitFixtureChannelIsReady(_: ARTRealtime) async throws {
 
 func waitForMapKeyUpdate(_ map: any LiveMap, _ key: String) async {
     await withCheckedContinuation { (continuation: CheckedContinuation<Void, _>) in
-        var subscription: SubscribeResponse!
-        subscription = map.subscribe { update in
+        map.subscribe { update, subscription in
             if update.update[key] != nil {
                 subscription.unsubscribe()
                 continuation.resume()
@@ -102,8 +101,7 @@ func waitForMapKeyUpdate(_ map: any LiveMap, _ key: String) async {
 
 func waitForCounterUpdate(_ counter: any LiveCounter) async {
     await withCheckedContinuation { (continuation: CheckedContinuation<Void, _>) in
-        var subscription: SubscribeResponse!
-        subscription = counter.subscribe { _ in
+        counter.subscribe { _, subscription in
             subscription.unsubscribe()
             continuation.resume()
         }
@@ -642,7 +640,7 @@ private struct ObjectsIntegrationTests {
 
                         async let counterSubPromise: Void = withCheckedThrowingContinuation { continuation in
                             do {
-                                try #require(root.get(key: "counter")?.liveCounterValue).subscribe { update in
+                                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()
                                 }