fix(realtime): auto reconnect after calling disconnect, and several refactors (supabase#627)

* fix: trigger socket reconnection on a new Task

* use factory function for returning ws connection

* code format

* use FakeWebSocket on tests

* integration tests

* fix FakeWebSocket

* add docs for EventEmitter

* use SupabaseClient for integration test

* fire channel rejoins on background

* allow multiple channels to the same topic

* remove Socket abstraction and use client directly

* skip linux CI

* fix build

Author: Guilherme Souza

.github/workflows/ci.yml

@@ -78,29 +78,29 @@ jobs:
         if: matrix.skip_release != '1'
         run: make XCODEBUILD_ARGUMENT="${{ matrix.command }}" CONFIG=Release PLATFORM="${{ matrix.platform }}" xcodebuild
 
-  linux:
-    name: linux
-    strategy:
-      matrix:
-        swift-version: ["5.10"]
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - uses: swift-actions/setup-swift@v2
-        with:
-          swift-version: ${{ matrix.swift-version }}
-      - name: Cache build
-        uses: actions/cache@v3
-        with:
-          path: |
-            .build
-          key: |
-            build-spm-linux-${{ matrix.swift-version }}-${{ hashFiles('**/Sources/**/*.swift', '**/Tests/**/*.swift', '**/Package.resolved') }}
-          restore-keys: |
-            build-spm-linux-${{ matrix.swift-version }}-
-      - run: make dot-env
-      - name: Run tests
-        run: swift test --skip IntegrationTests
+  # linux:
+  #   name: linux
+  #   strategy:
+  #     matrix:
+  #       swift-version: ["5.10"]
+  #   runs-on: ubuntu-latest
+  #   steps:
+  #     - uses: actions/checkout@v4
+  #     - uses: swift-actions/setup-swift@v2
+  #       with:
+  #         swift-version: ${{ matrix.swift-version }}
+  #     - name: Cache build
+  #       uses: actions/cache@v3
+  #       with:
+  #         path: |
+  #           .build
+  #         key: |
+  #           build-spm-linux-${{ matrix.swift-version }}-${{ hashFiles('**/Sources/**/*.swift', '**/Tests/**/*.swift', '**/Package.resolved') }}
+  #         restore-keys: |
+  #           build-spm-linux-${{ matrix.swift-version }}-
+  #     - run: make dot-env
+  #     - name: Run tests
+  #       run: swift test --skip IntegrationTests
 
   # library-evolution:
   #   name: Library (evolution)

Package.swift

@@ -92,10 +92,7 @@ let package = Package(
         .product(name: "InlineSnapshotTesting", package: "swift-snapshot-testing"),
         .product(name: "XCTestDynamicOverlay", package: "xctest-dynamic-overlay"),
         "Helpers",
-        "Auth",
-        "PostgREST",
-        "Realtime",
-        "Storage",
+        "Supabase",
         "TestHelpers",
       ],
       resources: [.process("Fixtures")]

Sources/Helpers/EventEmitter.swift

@@ -8,6 +8,9 @@
 import ConcurrencyExtras
 import Foundation
 
+/// A token for cancelling observations.
+///
+/// When this token gets deallocated it cancels the observation it was associated with. Store this token in another object to keep the observation alive.
 public final class ObservationToken: @unchecked Sendable, Hashable {
   private let _isCancelled = LockIsolated(false)
   package var onCancel: @Sendable () -> Void
@@ -44,9 +47,7 @@ public final class ObservationToken: @unchecked Sendable, Hashable {
   public func hash(into hasher: inout Hasher) {
     hasher.combine(ObjectIdentifier(self))
   }
-}
 
-extension ObservationToken {
   public func store(in collection: inout some RangeReplaceableCollection<ObservationToken>) {
     collection.append(self)
   }
@@ -59,20 +60,29 @@ extension ObservationToken {
 package final class EventEmitter<Event: Sendable>: Sendable {
   public typealias Listener = @Sendable (Event) -> Void
 
-  private let listeners = LockIsolated<[(key: ObjectIdentifier, listener: Listener)]>([])
-  private let _lastEvent: LockIsolated<Event>
-  package var lastEvent: Event { _lastEvent.value }
+  struct MutableState {
+    var listeners: [(key: ObjectIdentifier, listener: Listener)] = []
+    var lastEvent: Event
+  }
+
+  let mutableState: LockIsolated<MutableState>
+
+  /// The last event emitted by this Emiter, or the initial event.
+  package var lastEvent: Event { mutableState.lastEvent }
 
   let emitsLastEventWhenAttaching: Bool
 
   package init(
     initialEvent event: Event,
     emitsLastEventWhenAttaching: Bool = true
   ) {
-    _lastEvent = LockIsolated(event)
+    mutableState = LockIsolated(MutableState(lastEvent: event))
     self.emitsLastEventWhenAttaching = emitsLastEventWhenAttaching
   }
 
+  /// Attaches a new listener for observing event emissions.
+  ///
+  /// If emitter initialized with `emitsLastEventWhenAttaching = true`, listener gets called right away with last event.
   package func attach(_ listener: @escaping Listener) -> ObservationToken {
     defer {
       if emitsLastEventWhenAttaching {
@@ -84,21 +94,24 @@ package final class EventEmitter<Event: Sendable>: Sendable {
     let key = ObjectIdentifier(token)
 
     token.onCancel = { [weak self] in
-      self?.listeners.withValue {
-        $0.removeAll { $0.key == key }
+      self?.mutableState.withValue {
+        $0.listeners.removeAll { $0.key == key }
       }
     }
 
-    listeners.withValue {
-      $0.append((key, listener))
+    mutableState.withValue {
+      $0.listeners.append((key, listener))
     }
 
     return token
   }
 
+  /// Trigger a new event on all attached listeners, or a specific listener owned by the `token` provided.
   package func emit(_ event: Event, to token: ObservationToken? = nil) {
-    _lastEvent.setValue(event)
-    let listeners = listeners.value
+    let listeners = mutableState.withValue {
+      $0.lastEvent = event
+      return $0.listeners
+    }
 
     if let token {
       listeners.first { $0.key == ObjectIdentifier(token) }?.listener(event)
@@ -109,6 +122,7 @@ package final class EventEmitter<Event: Sendable>: Sendable {
     }
   }
 
+  /// Returns a new ``AsyncStream`` for observing events emitted by this emitter.
   package func stream() -> AsyncStream<Event> {
     AsyncStream { continuation in
       let token = attach { status in

Sources/Realtime/V2/PushV2.swift

@@ -31,7 +31,7 @@ actor PushV2 {
       return .error
     }
 
-    await channel.socket.push(message)
+    channel.socket.push(message)
 
     if !channel.config.broadcast.acknowledgeBroadcasts {
       // channel was configured with `ack = false`,
@@ -40,7 +40,7 @@ actor PushV2 {
     }
 
     do {
-      return try await withTimeout(interval: channel.socket.options().timeoutInterval) {
+      return try await withTimeout(interval: channel.socket.options.timeoutInterval) {
         await withCheckedContinuation { continuation in
           self.receivedContinuation = continuation
         }

Sources/Realtime/V2/RealtimeChannelV2.swift

@@ -25,46 +25,6 @@ public struct RealtimeChannelConfig: Sendable {
   public var isPrivate: Bool
 }
 
-struct Socket: Sendable {
-  var broadcastURL: @Sendable () -> URL
-  var status: @Sendable () -> RealtimeClientStatus
-  var options: @Sendable () -> RealtimeClientOptions
-  var accessToken: @Sendable () async -> String?
-  var apiKey: @Sendable () -> String?
-  var makeRef: @Sendable () -> Int
-
-  var connect: @Sendable () async -> Void
-  var addChannel: @Sendable (_ channel: RealtimeChannelV2) -> Void
-  var removeChannel: @Sendable (_ channel: RealtimeChannelV2) async -> Void
-  var push: @Sendable (_ message: RealtimeMessageV2) async -> Void
-  var httpSend: @Sendable (_ request: Helpers.HTTPRequest) async throws -> Helpers.HTTPResponse
-}
-
-extension Socket {
-  init(client: RealtimeClientV2) {
-    self.init(
-      broadcastURL: { [weak client] in client?.broadcastURL ?? URL(string: "http://localhost")! },
-      status: { [weak client] in client?.status ?? .disconnected },
-      options: { [weak client] in client?.options ?? .init() },
-      accessToken: { [weak client] in
-        if let accessToken = try? await client?.options.accessToken?() {
-          return accessToken
-        }
-        return client?.mutableState.accessToken
-      },
-      apiKey: { [weak client] in client?.apikey },
-      makeRef: { [weak client] in client?.makeRef() ?? 0 },
-      connect: { [weak client] in await client?.connect() },
-      addChannel: { [weak client] in client?.addChannel($0) },
-      removeChannel: { [weak client] in await client?.removeChannel($0) },
-      push: { [weak client] in await client?.push($0) },
-      httpSend: { [weak client] in
-        try await client?.http.send($0) ?? .init(data: Data(), response: HTTPURLResponse())
-      }
-    )
-  }
-}
-
 public final class RealtimeChannelV2: Sendable {
   struct MutableState {
     var clientChanges: [PostgresJoinConfig] = []
@@ -77,7 +37,8 @@ public final class RealtimeChannelV2: Sendable {
   let topic: String
   let config: RealtimeChannelConfig
   let logger: (any SupabaseLogger)?
-  let socket: Socket
+  let socket: RealtimeClientV2
+  var joinRef: String? { mutableState.joinRef }
 
   let callbackManager = CallbackManager()
   private let statusEventEmitter = EventEmitter<RealtimeChannelStatus>(initialEvent: .unsubscribed)
@@ -105,7 +66,7 @@ public final class RealtimeChannelV2: Sendable {
   init(
     topic: String,
     config: RealtimeChannelConfig,
-    socket: Socket,
+    socket: RealtimeClientV2,
     logger: (any SupabaseLogger)?
   ) {
     self.topic = topic
@@ -120,8 +81,8 @@ public final class RealtimeChannelV2: Sendable {
 
   /// Subscribes to the channel
   public func subscribe() async {
-    if socket.status() != .connected {
-      if socket.options().connectOnSubscribe != true {
+    if socket.status != .connected {
+      if socket.options.connectOnSubscribe != true {
         reportIssue(
           "You can't subscribe to a channel while the realtime client is not connected. Did you forget to call `realtime.connect()`?"
         )
@@ -130,8 +91,6 @@ public final class RealtimeChannelV2: Sendable {
       await socket.connect()
     }
 
-    socket.addChannel(self)
-
     status = .subscribing
     logger?.debug("Subscribing to channel \(topic)")
 
@@ -144,10 +103,10 @@ public final class RealtimeChannelV2: Sendable {
 
     let payload = RealtimeJoinPayload(
       config: joinConfig,
-      accessToken: await socket.accessToken()
+      accessToken: await socket._getAccessToken()
     )
 
-    let joinRef = socket.makeRef().description
+    let joinRef = socket.makeRef()
     mutableState.withValue { $0.joinRef = joinRef }
 
     logger?.debug("Subscribing to channel with body: \(joinConfig)")
@@ -159,7 +118,7 @@ public final class RealtimeChannelV2: Sendable {
     )
 
     do {
-      try await withTimeout(interval: socket.options().timeoutInterval) { [self] in
+      try await withTimeout(interval: socket.options.timeoutInterval) { [self] in
         _ = await statusChange.first { @Sendable in $0 == .subscribed }
       }
     } catch {
@@ -215,17 +174,17 @@ public final class RealtimeChannelV2: Sendable {
       }
 
       var headers: HTTPFields = [.contentType: "application/json"]
-      if let apiKey = socket.apiKey() {
+      if let apiKey = socket.options.apikey {
         headers[.apiKey] = apiKey
       }
-      if let accessToken = await socket.accessToken() {
+      if let accessToken = await socket._getAccessToken() {
         headers[.authorization] = "Bearer \(accessToken)"
       }
 
       let task = Task { [headers] in
-        _ = try? await socket.httpSend(
+        _ = try? await socket.http.send(
           HTTPRequest(
-            url: socket.broadcastURL(),
+            url: socket.broadcastURL,
             method: .post,
             headers: headers,
             body: JSONEncoder().encode(
@@ -245,7 +204,7 @@ public final class RealtimeChannelV2: Sendable {
       }
 
       if config.broadcast.acknowledgeBroadcasts {
-        try? await withTimeout(interval: socket.options().timeoutInterval) {
+        try? await withTimeout(interval: socket.options.timeoutInterval) {
           await task.value
         }
       }
@@ -406,7 +365,7 @@ public final class RealtimeChannelV2: Sendable {
         callbackManager.triggerBroadcast(event: event, json: payload)
 
       case .close:
-        await socket.removeChannel(self)
+        socket._remove(self)
         logger?.debug("Unsubscribed from channel \(message.topic)")
         status = .unsubscribed
 
@@ -582,7 +541,7 @@ public final class RealtimeChannelV2: Sendable {
     let push = mutableState.withValue {
       let message = RealtimeMessageV2(
         joinRef: $0.joinRef,
-        ref: ref ?? socket.makeRef().description,
+        ref: ref ?? socket.makeRef(),
         topic: self.topic,
         event: event,
         payload: payload