feat(realtime): add callback for handling system events

Author: Guilherme Souza

Sources/Realtime/RealtimeChannel+AsyncAwait.swift

@@ -104,6 +104,21 @@ extension RealtimeChannelV2 {
 
     return stream
   }
+  
+  /// Listen for `system` event.
+  public func system() -> AsyncStream<RealtimeMessageV2> {
+    let (stream, continuation) = AsyncStream<RealtimeMessageV2>.makeStream()
+
+    let subscription = onSystem {
+      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:)")

Sources/Realtime/V2/CallbackManager.swift

@@ -75,6 +75,15 @@ final class CallbackManager: Sendable {
     }
   }
 
+  @discardableResult
+  func addSystemCallback(callback: @escaping @Sendable (RealtimeMessageV2) -> Void) -> Int {
+    mutableState.withValue {
+      $0.id += 1
+      $0.callbacks.append(.system(SystemCallback(id: $0.id, callback: callback)))
+      return $0.id
+    }
+  }
+
   func setServerChanges(changes: [PostgresJoinConfig]) {
     mutableState.withValue {
       $0.serverChanges = changes
@@ -167,16 +176,23 @@ struct PresenceCallback {
   var callback: @Sendable (any PresenceAction) -> Void
 }
 
+struct SystemCallback {
+  var id: Int
+  var callback: @Sendable (RealtimeMessageV2) -> Void
+}
+
 enum RealtimeCallback {
   case postgres(PostgresCallback)
   case broadcast(BroadcastCallback)
   case presence(PresenceCallback)
+  case system(SystemCallback)
 
   var id: Int {
     switch self {
     case let .postgres(callback): callback.id
     case let .broadcast(callback): callback.id
     case let .presence(callback): callback.id
+    case let .system(callback): callback.id
     }
   }
 }

Sources/Realtime/V2/RealtimeChannelV2.swift

@@ -545,6 +545,14 @@ public final class RealtimeChannelV2: Sendable {
     }
   }
 
+  public func onSystem(callback: @escaping @Sendable (RealtimeMessageV2) -> Void) -> Subscription {
+    let id = callbackManager.addSystemCallback(callback: callback)
+    return Subscription { [weak callbackManager, logger] in
+      logger?.debug("Removing system callback with id: \(id)")
+      callbackManager?.removeCallback(id: id)
+    }
+  }
+
   @discardableResult
   func push(_ event: String, ref: String? = nil, payload: JSONObject = [:]) async -> PushStatus {
     let push = mutableState.withValue {