feat(realtime): add closure based method for broadcast

Author: grdsdev

Sources/Realtime/RealtimeChannel+AsyncAwait.swift

@@ -101,4 +101,25 @@ extension RealtimeChannelV2 {
     }
     return stream
   }
+
+  /// Listen for broadcast messages sent by other clients within the same channel under a specific `event`.
+  public func broadcastStream(event: String) -> AsyncStream<JSONObject> {
+    let (stream, continuation) = AsyncStream<JSONObject>.makeStream()
+
+    let subscription = onBroadcast(event: event) {
+      continuation.yield($0)
+    }
+
+    continuation.onTermination = { _ in
+      subscription.cancel()
+    }
+
+    return stream
+  }
+
+  /// Listen for broadcast messages sent by other clients within the same channel under a specific `event`.
+  @available(*, deprecated, renamed: "broadcastStream(event:)")
+  public func broadcast(event: String) -> AsyncStream<JSONObject> {
+    broadcastStream(event: event)
+  }
 }

Sources/Realtime/V2/RealtimeChannelV2.swift

@@ -437,30 +437,16 @@ public actor RealtimeChannelV2 {
     }
   }
 
-  /// Listen for broadcast messages sent by other clients within the same channel under a specific
-  /// `event`.
-  public func broadcastStream(event: String) -> AsyncStream<JSONObject> {
-    let (stream, continuation) = AsyncStream<JSONObject>.makeStream()
-
-    let id = callbackManager.addBroadcastCallback(event: event) {
-      continuation.yield($0)
-    }
-
-    let logger = logger
-
-    continuation.onTermination = { [weak callbackManager] _ in
+  /// Listen for broadcast messages sent by other clients within the same channel under a specific `event`.
+  public func onBroadcast(
+    event: String,
+    callback: @escaping @Sendable (JSONObject) -> Void
+  ) -> Subscription {
+    let id = callbackManager.addBroadcastCallback(event: event, callback: callback)
+    return Subscription { [weak callbackManager, logger] in
       logger?.debug("Removing broadcast callback with id: \(id)")
       callbackManager?.removeCallback(id: id)
     }
-
-    return stream
-  }
-
-  /// Listen for broadcast messages sent by other clients within the same channel under a specific
-  /// `event`.
-  @available(*, deprecated, renamed: "broadcastStream(event:)")
-  public func broadcast(event: String) -> AsyncStream<JSONObject> {
-    broadcastStream(event: event)
   }
 
   @discardableResult

Sources/_Helpers/EventEmitter.swift

@@ -12,7 +12,7 @@ public final class ObservationToken: Sendable {
   let _onCancel = LockIsolated((@Sendable () -> Void)?.none)
 
   package init(_ onCancel: (@Sendable () -> Void)? = nil) {
-    self._onCancel.setValue(onCancel)
+    _onCancel.setValue(onCancel)
   }
 
   @available(*, deprecated, renamed: "cancel")