test: add test for channel subscribe timeout

Author: Guilherme Souza

Sources/Realtime/V2/PushV2.swift

@@ -30,9 +30,10 @@ actor PushV2 {
           }
         }
       } catch is TimeoutError {
+        channel?.logger?.debug("Push timed out.")
         return .timeout
       } catch {
-        channel?.logger?.error("error sending Push: \(error)")
+        channel?.logger?.error("Error sending push: \(error)")
         return .error
       }
     }

Sources/Realtime/V2/RealtimeChannelV2.swift

@@ -106,7 +106,18 @@ public actor RealtimeChannelV2 {
       )
     )
 
-    _ = await statusChange.first { @Sendable in $0 == .subscribed }
+    do {
+      try await withTimeout(interval: socket?.config.timeoutInterval ?? 10) { [self] in
+        _ = await statusChange.first { @Sendable in $0 == .subscribed }
+      }
+    } catch {
+      if error is TimeoutError {
+        logger?.debug("subscribe timed out.")
+        await subscribe()
+      } else {
+        logger?.error("subscribe failed: \(error)")
+      }
+    }
   }
 
   public func unsubscribe() async {

Tests/RealtimeTests/RealtimeTests.swift

@@ -27,7 +27,9 @@ final class RealtimeTests: XCTestCase {
         url: url,
         apiKey: apiKey,
         heartbeatInterval: 1,
-        reconnectDelay: 1
+        reconnectDelay: 1,
+        timeoutInterval: 2,
+        logger: TestLogger()
       ),
       ws: ws
     )
@@ -62,10 +64,85 @@ final class RealtimeTests: XCTestCase {
       // Wait until channel subscribed
       await subscription.value
 
-      XCTAssertNoDifference(ws.sentMessages.value, [.subscribeToMessages])
+      XCTAssertNoDifference(ws.sentMessages.value, [.subscribeToMessages(ref: "1", joinRef: "1")])
     }
   }
 
+  func testSubscribeTimeout() async throws {
+    let channel = await sut.channel("public:messages")
+    let joinEventCount = LockIsolated(0)
+
+    ws.on { message in
+      if message.event == "heartbeat" {
+        return RealtimeMessageV2(
+          joinRef: message.joinRef,
+          ref: message.ref,
+          topic: "phoenix",
+          event: "phx_reply",
+          payload: [
+            "response": [:],
+            "status": "ok",
+          ]
+        )
+      }
+
+      if message.event == "phx_join" {
+        joinEventCount.withValue { $0 += 1 }
+
+        // Skip first join.
+        if joinEventCount.value == 2 {
+          return .messagesSubscribed
+        }
+      }
+
+      return nil
+    }
+
+    await connectSocketAndWait()
+
+    Task {
+      await channel.subscribe()
+    }
+
+    await Task.megaYield()
+
+    try? await Task.sleep(nanoseconds: NSEC_PER_SEC * 2)
+
+    let joinSentMessages = ws.sentMessages.value.filter { $0.event == "phx_join" }
+
+    let expectedMessages = try [
+      RealtimeMessageV2(
+        joinRef: "1",
+        ref: "1",
+        topic: "realtime:public:messages",
+        event: "phx_join",
+        payload: JSONObject(
+          RealtimeJoinPayload(
+            config: RealtimeJoinConfig(),
+            accessToken: apiKey
+          )
+        )
+      ),
+      RealtimeMessageV2(
+        joinRef: "3",
+        ref: "3",
+        topic: "realtime:public:messages",
+        event: "phx_join",
+        payload: JSONObject(
+          RealtimeJoinPayload(
+            config: RealtimeJoinConfig(),
+            accessToken: apiKey
+          )
+        )
+      ),
+    ]
+
+    XCTAssertNoDifference(
+      joinSentMessages,
+      expectedMessages
+    )
+  }
+
   func testHeartbeat() async throws {
     try await withTimeout(interval: 4) { [self] in
       let expectation = expectation(description: "heartbeat")
@@ -155,27 +232,29 @@ final class RealtimeTests: XCTestCase {
 }
 
 extension RealtimeMessageV2 {
-  static let subscribeToMessages = Self(
-    joinRef: "1",
-    ref: "1",
-    topic: "realtime:public:messages",
-    event: "phx_join",
-    payload: [
-      "access_token": "anon.api.key",
-      "config": [
-        "broadcast": [
-          "self": false,
-          "ack": false,
-        ],
-        "postgres_changes": [
-          ["table": "messages", "event": "INSERT", "schema": "public"],
-          ["table": "messages", "schema": "public", "event": "UPDATE"],
-          ["schema": "public", "table": "messages", "event": "DELETE"],
+  static func subscribeToMessages(ref: String?, joinRef: String?) -> RealtimeMessageV2 {
+    Self(
+      joinRef: joinRef,
+      ref: ref,
+      topic: "realtime:public:messages",
+      event: "phx_join",
+      payload: [
+        "access_token": "anon.api.key",
+        "config": [
+          "broadcast": [
+            "self": false,
+            "ack": false,
+          ],
+          "postgres_changes": [
+            ["table": "messages", "event": "INSERT", "schema": "public"],
+            ["table": "messages", "schema": "public", "event": "UPDATE"],
+            ["schema": "public", "table": "messages", "event": "DELETE"],
+          ],
+          "presence": ["key": ""],
         ],
-        "presence": ["key": ""],
-      ],
-    ]
-  )
+      ]
+    )
+  }
 
   static let messagesSubscribed = Self(
     joinRef: nil,
@@ -205,3 +284,9 @@ extension RealtimeMessageV2 {
     ]
   )
 }
+
+struct TestLogger: SupabaseLogger {
+  func log(message: SupabaseLogMessage) {
+    print(message.description)
+  }
+}