Introduction of isConnected Publisher (#47)

* Add isConnected publisher

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ jobs:
 
     steps:
       - uses: actions/checkout@v3
-      - name: Select Xcode 14
-        run: sudo xcode-select -s /Applications/Xcode_14.3.1.app
+      - name: Select Xcode 15
+        run: sudo xcode-select -s /Applications/Xcode_15.0.app
       - name: Test
         run: swift test

.swift-version

@@ -1,2 +1,2 @@
-5.8.0
+5.9.0
 

Package.resolved

@@ -5,8 +5,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/shareup/async-extensions.git",
       "state" : {
-        "revision" : "f8dec7a227bbbe15fd4df90c787d4c73e91451ba",
-        "version" : "4.2.1"
+        "revision" : "7e727e3b9009a5de429393691f9f499aedb7a109",
+        "version" : "4.3.0"
       }
     },
     {
@@ -50,8 +50,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/shareup/websocket-apple.git",
       "state" : {
-        "revision" : "f409dca92eb61be6dd1183941a8590f9c4459f07",
-        "version" : "4.0.2"
+        "revision" : "32de0d7733ded57f1b052ab5dddf36708b0ab1dc",
+        "version" : "4.0.3"
       }
     }
   ],

Package.swift

@@ -1,4 +1,4 @@
-// swift-tools-version:5.8
+// swift-tools-version:5.9
 import PackageDescription
 
 let package = Package(
@@ -12,7 +12,7 @@ let package = Package(
     dependencies: [
         .package(
             url: "https://github.com/shareup/async-extensions.git",
-            from: "4.1.0"
+            from: "4.3.0"
         ),
         .package(
             url: "https://github.com/shareup/dispatch-timer.git",
@@ -32,7 +32,7 @@ let package = Package(
         ),
         .package(
             url: "https://github.com/shareup/websocket-apple.git",
-            from: "4.0.2"
+            from: "4.0.3"
         ),
     ],
     targets: [

Sources/Phoenix/Event.swift

@@ -33,19 +33,19 @@ public enum Event: Hashable, ExpressibleByStringLiteral, Sendable {
     public var stringValue: String {
         switch self {
         case .join:
-            return "phx_join"
+            "phx_join"
         case .leave:
-            return "phx_leave"
+            "phx_leave"
         case .close:
-            return "phx_close"
+            "phx_close"
         case .reply:
-            return "phx_reply"
+            "phx_reply"
         case .error:
-            return "phx_error"
+            "phx_error"
         case .heartbeat:
-            return "heartbeat"
+            "heartbeat"
         case let .custom(string):
-            return string
+            string
         }
     }
 

Sources/Phoenix/PhoenixChannel.swift

@@ -14,7 +14,9 @@ final class PhoenixChannel: @unchecked Sendable {
     let topic: Topic
     let joinPayload: JSON
 
-    var messages: AsyncStream<Message> { messageSubject.allValues }
+    var messages: AsyncStream<Message> {
+        messageSubject.allAsyncValues
+    }
 
     var isErrored: Bool { state.access { $0.isErrored } }
     var isJoining: Bool { state.access { $0.isJoining } }
@@ -167,7 +169,7 @@ final class PhoenixChannel: @unchecked Sendable {
 private extension PhoenixChannel {
     func watchSocketConnection() {
         tasks.storedTask(key: "socketConnection") { [socket, state, weak self] in
-            let connectionStates = socket.onConnectionStateChange.allValues
+            let connectionStates = socket.onConnectionStateChange.allAsyncValues
             for await connectionState in connectionStates {
                 try Task.checkCancellation()
 
@@ -254,13 +256,13 @@ private struct State: @unchecked Sendable {
     var joinRef: Ref? {
         switch connection {
         case .unjoined, .errored, .joining, .left:
-            return nil
+            nil
 
         case let .joined(ref, _):
-            return ref
+            ref
 
         case let .leaving(ref, _):
-            return ref
+            ref
         }
     }
 
@@ -294,12 +296,12 @@ private struct State: @unchecked Sendable {
 
         var description: String {
             switch self {
-            case .errored: return "errored"
-            case .joined: return "joined"
-            case .joining: return "joining"
-            case .leaving: return "leaving"
-            case .left: return "left"
-            case .unjoined: return "unjoined"
+            case .errored: "errored"
+            case .joined: "joined"
+            case .joining: "joining"
+            case .leaving: "leaving"
+            case .left: "left"
+            case .unjoined: "unjoined"
             }
         }
     }
@@ -495,9 +497,9 @@ private struct State: @unchecked Sendable {
             switch connection {
             case let .joined(joinRef, _):
                 if message.joinRef == joinRef || message.joinRef == nil {
-                    return { sendMessage(message) }
+                    { sendMessage(message) }
                 } else {
-                    return {
+                    {
                         os_log(
                             "outdated message: channel=%{public}s joinRef=%d message=%d",
                             log: .phoenix,
@@ -510,7 +512,7 @@ private struct State: @unchecked Sendable {
                 }
 
             case .unjoined, .errored, .joining, .leaving, .left:
-                return { sendMessage(message) }
+                { sendMessage(message) }
             }
         }
 

Sources/Phoenix/PhoenixSocket.swift

@@ -23,10 +23,10 @@ final actor PhoenixSocket {
     var webSocket: WebSocket? {
         switch _connectionState.value {
         case let .connecting(ws), let .open(ws), let .closing(ws):
-            return ws
+            ws
 
         case .waitingToReconnect, .preparingToReconnect, .closed:
-            return nil
+            nil
         }
     }
 
@@ -37,7 +37,10 @@ final actor PhoenixSocket {
     nonisolated let pushEncoder: PushEncoder
     nonisolated let messageDecoder: MessageDecoder
 
-    nonisolated var messages: AsyncStream<Message> { messageSubject.allValues }
+    nonisolated var messages: AsyncStream<Message> {
+        messageSubject.allAsyncValues
+    }
+
     private nonisolated let messageSubject = PassthroughSubject<Message, Never>()
 
     private nonisolated let currentWebSocketID = Locked(0)
@@ -51,6 +54,12 @@ final actor PhoenixSocket {
         _connectionState.eraseToAnyPublisher()
     }
 
+    nonisolated var isConnected: AnyPublisher<Bool, Never> {
+        _connectionState
+            .map(\.isOpen)
+            .eraseToAnyPublisher()
+    }
+
     private nonisolated let _connectionState =
         CurrentValueSubject<
             PhoenixSocket.ConnectionState,
@@ -230,7 +239,6 @@ extension PhoenixSocket {
                 guard let self else { return }
 
                 do {
-
                     let message = try decoder(msg)
 
                     _ = pushes.didReceive(message)
@@ -368,12 +376,12 @@ extension PhoenixSocket {
 
         var description: String {
             switch self {
-            case .closed: return "closed"
-            case .waitingToReconnect: return "waitingToReconnect"
-            case .preparingToReconnect: return "preparingToReconnect"
-            case .connecting: return "connecting"
-            case .open: return "open"
-            case .closing: return "closing"
+            case .closed: "closed"
+            case .waitingToReconnect: "waitingToReconnect"
+            case .preparingToReconnect: "preparingToReconnect"
+            case .connecting: "connecting"
+            case .open: "open"
+            case .closing: "closing"
             }
         }
     }

Sources/Phoenix/PushBuffer.swift

@@ -152,7 +152,7 @@ Sendable {
                             if cancellation.isCancelled {
                                 throw CancellationError()
                             } else {
-                                return try state.next(cont)
+                                try state.next(cont)
                             }
                         }
 

Sources/Phoenix/Socket.swift

@@ -1,3 +1,4 @@
+import Combine
 import Foundation
 import JSON
 import Synchronized
@@ -7,6 +8,7 @@ public struct Socket: Identifiable, Sendable {
     public let id: Int
     public var connect: @Sendable () async -> Void
     public var disconnect: @Sendable () async -> Void
+    public var isConnected: @Sendable () async -> AnyPublisher<Bool, Never>
     public var channel: @Sendable (Topic, JSON) async -> Channel
     public var remove: @Sendable (Topic) async -> Void
     public var removeAll: @Sendable () async -> Void
@@ -18,6 +20,7 @@ public struct Socket: Identifiable, Sendable {
         id: Int,
         connect: @escaping @Sendable () async -> Void,
         disconnect: @escaping @Sendable () async -> Void,
+        isConnected: @escaping @Sendable () async -> AnyPublisher<Bool, Never>,
         channel: @escaping @Sendable (Topic, JSON) async -> Channel,
         remove: @escaping @Sendable (Topic) async -> Void,
         removeAll: @escaping @Sendable () async -> Void,
@@ -28,6 +31,7 @@ public struct Socket: Identifiable, Sendable {
         self.id = id
         self.connect = connect
         self.disconnect = disconnect
+        self.isConnected = isConnected
         self.channel = channel
         self.remove = remove
         self.removeAll = removeAll
@@ -62,6 +66,7 @@ public extension Socket {
             id: nextSocketID(),
             connect: { await phoenix.connect() },
             disconnect: { await phoenix.disconnect() },
+            isConnected: { phoenix.isConnected },
             channel: { topic, joinPayload in
                 Channel.with(
                     await phoenix.channel(

Tests/PhoenixTests/PhoenixChannelTests.swift

@@ -1003,7 +1003,7 @@ private extension PhoenixChannelTests {
     }
 
     var outgoingMessages: AsyncStream<WebSocketMessage> {
-        sendSubject.allValues
+        sendSubject.allAsyncValues
     }
 
     func nextOutoingMessage() async throws -> Message {

Tests/PhoenixTests/PhoenixSocketTests.swift

@@ -214,6 +214,35 @@ final class PhoenixSocketTests: XCTestCase {
         }
     }
 
+    func testIsConnectedPublisher() async throws {
+        let didConnect = future(timeout: 2)
+        let didDisconnect = future(timeout: 2)
+
+        let socket = PhoenixSocket(
+            url: url,
+            makeWebSocket: makeFakeWebSocket
+        )
+
+        let sub = socket.isConnected
+            .sink { isConnected in
+                if isConnected {
+                    didConnect.resolve()
+                } else {
+                    didDisconnect.resolve()
+                }
+            }
+
+        await socket.connect()
+
+        try await didConnect.value
+
+        await socket.disconnect()
+
+        try await didDisconnect.value
+
+        sub.cancel()
+    }
+
     // MARK: "channel"
 
     func testCreateChannelWithTopicAndPayload() async throws {
@@ -768,7 +797,7 @@ private extension PhoenixSocketTests {
     var makeFakeWebSocket: MakeWebSocket {
         { [weak self] _, _, _, onOpen, onClose async throws in
             guard let self else { throw CancellationError() }
-            return self.fake(onOpen: onOpen, onClose: onClose)
+            return fake(onOpen: onOpen, onClose: onClose)
         }
     }
 
@@ -820,7 +849,7 @@ private extension PhoenixSocketTests {
     }
 
     var outgoingMessages: AsyncStream<WebSocketMessage> {
-        sendSubject.allValues
+        sendSubject.allAsyncValues
     }
 
     func sendReply(