Merge pull request #8 from GraphQLSwift/feat/pass-init-result

Major cleanup

Author: NeedleInAJayStack

README.md

@@ -1,5 +1,8 @@
 # GraphQLWS
 
+[![](https://img.shields.io/endpoint?url=https%3A%2F%2Fswiftpackageindex.com%2Fapi%2Fpackages%2FGraphQLSwift%2FGraphQLWS%2Fbadge%3Ftype%3Dplatforms)](https://swiftpackageindex.com/GraphQLSwift/GraphQLWS)
+[![](https://img.shields.io/endpoint?url=https%3A%2F%2Fswiftpackageindex.com%2Fapi%2Fpackages%2FGraphQLSwift%2FGraphQLWS%2Fbadge%3Ftype%3Dswift-versions)](https://swiftpackageindex.com/GraphQLSwift/GraphQLWS)
+
 This implements the [graphql-ws WebSocket subprotocol](https://github.com/apollographql/subscriptions-transport-ws/blob/master/PROTOCOL.md).
 It is mainly intended for server support, but there is a basic client implementation included.
 
@@ -14,7 +17,7 @@ Features:
 To use this package, include it in your `Package.swift` dependencies:
 
 ```swift
-.package(url: "git@gitlab.com:PassiveLogic/platform/GraphQLWS.git", from: "<version>"),
+.package(url: "https://github.com/GraphQLSwift/GraphQLWS", from: "<version>"),
 ```
 
 Then create a class to implement the `Messenger` protocol. Here's an example using
@@ -25,33 +28,18 @@ import WebSocketKit
 import GraphQLWS
 
 /// Messenger wrapper for WebSockets
-class WebSocketMessenger: Messenger {
-    private weak var websocket: WebSocket?
-    private var onReceive: (String) -> Void = { _ in }
-
-    init(websocket: WebSocket) {
-        self.websocket = websocket
-        websocket.onText { _, message in
-            try await self.onReceive(message)
-        }
-    }
+struct WebSocketMessenger: Messenger {
+    let websocket: WebSocket
 
     func send<S>(_ message: S) async throws where S: Collection, S.Element == Character async throws {
-        guard let websocket = websocket else { return }
         try await websocket.send(message)
     }
 
-    func onReceive(callback: @escaping (String) async throws -> Void) {
-        self.onReceive = callback
-    }
-
     func error(_ message: String, code: Int) async throws {
-        guard let websocket = websocket else { return }
         try await websocket.send("\(code): \(message)")
     }
 
     func close() async throws {
-        guard let websocket = websocket else { return }
         try await websocket.close()
     }
 }
@@ -85,6 +73,12 @@ routes.webSocket(
                 )
             }
         )
+        let incoming = AsyncStream<String> { continuation in
+            websocket.onText { _, message in
+                continuation.yield(message)
+            }
+        }
+        try await server.listen(to: incoming)
     }
 )
 ```
@@ -125,12 +119,3 @@ This example would require `connection_init` message from the client to look lik
 ```
 
 If the `payload` field is not required on your server, you may make Server's generic declaration optional like `Server<Payload?>`
-
-## Memory Management
-
-Memory ownership among the Server, Client, and Messenger may seem a little backwards. This is because the Swift/Vapor WebSocket
-implementation persists WebSocket objects long after their callback and they are expected to retain strong memory references to the
-objects required for responses. In order to align cleanly and avoid memory cycles, Server and Client are injected strongly into Messenger
-callbacks, and only hold weak references to their Messenger. This means that Messenger objects (or their enclosing WebSocket) must
-be persisted to have the connected Server or Client objects function. That is, if a Server's Messenger falls out of scope and deinitializes,
-the Server will no longer respond to messages.

Sources/GraphQLWS/Client.swift

@@ -2,17 +2,15 @@ import Foundation
 import GraphQL
 
 /// Client is an open-ended implementation of the client side of the protocol. It parses and adds callbacks for each type of server respose.
-public class Client<InitPayload: Equatable & Codable> {
-    // We keep this weak because we strongly inject this object into the messenger callback
-    weak var messenger: Messenger?
+public actor Client<InitPayload: Equatable & Codable> {
+    let messenger: Messenger
 
-    var onConnectionError: (ConnectionErrorResponse, Client) async throws -> Void = { _, _ in }
-    var onConnectionAck: (ConnectionAckResponse, Client) async throws -> Void = { _, _ in }
-    var onConnectionKeepAlive: (ConnectionKeepAliveResponse, Client) async throws -> Void = { _, _ in }
-    var onData: (DataResponse, Client) async throws -> Void = { _, _ in }
-    var onError: (ErrorResponse, Client) async throws -> Void = { _, _ in }
-    var onComplete: (CompleteResponse, Client) async throws -> Void = { _, _ in }
-    var onMessage: (String, Client) async throws -> Void = { _, _ in }
+    let onConnectionError: (ConnectionErrorResponse, Client) async throws -> Void
+    let onConnectionAck: (ConnectionAckResponse, Client) async throws -> Void
+    let onConnectionKeepAlive: (ConnectionKeepAliveResponse, Client) async throws -> Void
+    let onData: (DataResponse, Client) async throws -> Void
+    let onError: (ErrorResponse, Client) async throws -> Void
+    let onComplete: (CompleteResponse, Client) async throws -> Void
 
     let encoder = GraphQLJSONEncoder()
     let decoder = JSONDecoder()
@@ -21,120 +19,98 @@ public class Client<InitPayload: Equatable & Codable> {
     ///
     /// - Parameters:
     ///   - messenger: The messenger to bind the client to.
+    ///   - onConnectionError: The callback run on receipt of a `connection_error` message
+    ///   - onConnectionAck: The callback run on receipt of a `connection_ack` message
+    ///   - onConnectionKeepAlive: The callback run on receipt of a `connection_ka` message
+    ///   - onData: The callback run on receipt of a `data` message
+    ///   - onError: The callback run on receipt of an `error` message
+    ///   - onComplete: The callback run on receipt of a `complete` message
     public init(
-        messenger: Messenger
+        messenger: Messenger,
+        onConnectionError: @escaping (ConnectionErrorResponse, Client) async throws -> Void = { _, _ in },
+        onConnectionAck: @escaping (ConnectionAckResponse, Client) async throws -> Void = { _, _ in },
+        onConnectionKeepAlive: @escaping (ConnectionKeepAliveResponse, Client) async throws -> Void = { _, _ in },
+        onData: @escaping (DataResponse, Client) async throws -> Void = { _, _ in },
+        onError: @escaping (ErrorResponse, Client) async throws -> Void = { _, _ in },
+        onComplete: @escaping (CompleteResponse, Client) async throws -> Void = { _, _ in }
     ) {
         self.messenger = messenger
-        messenger.onReceive { message in
-            try await self.onMessage(message, self)
+        self.onConnectionError = onConnectionError
+        self.onConnectionAck = onConnectionAck
+        self.onConnectionKeepAlive = onConnectionKeepAlive
+        self.onData = onData
+        self.onError = onError
+        self.onComplete = onComplete
+    }
 
+    /// Listen and react to the provided async sequence of server messages. This function will block until the stream is completed.
+    /// - Parameter incoming: The server message sequence that the client should react to.
+    public func listen<A: AsyncSequence & Sendable>(to incoming: A) async throws -> Void where A.Element == String {
+        for try await message in incoming {
             // Detect and ignore error responses.
             if message.starts(with: "44") {
                 // TODO: Determine what to do with returned error messages
                 return
             }
 
             guard let json = message.data(using: .utf8) else {
-                try await self.error(.invalidEncoding())
+                try await error(.invalidEncoding())
                 return
             }
 
             let response: Response
             do {
-                response = try self.decoder.decode(Response.self, from: json)
+                response = try decoder.decode(Response.self, from: json)
             } catch {
                 try await self.error(.noType())
                 return
             }
 
             switch response.type {
             case .GQL_CONNECTION_ERROR:
-                guard let connectionErrorResponse = try? self.decoder.decode(ConnectionErrorResponse.self, from: json) else {
-                    try await self.error(.invalidResponseFormat(messageType: .GQL_CONNECTION_ERROR))
+                guard let connectionErrorResponse = try? decoder.decode(ConnectionErrorResponse.self, from: json) else {
+                    try await error(.invalidResponseFormat(messageType: .GQL_CONNECTION_ERROR))
                     return
                 }
-                try await self.onConnectionError(connectionErrorResponse, self)
+                try await onConnectionError(connectionErrorResponse, self)
             case .GQL_CONNECTION_ACK:
-                guard let connectionAckResponse = try? self.decoder.decode(ConnectionAckResponse.self, from: json) else {
-                    try await self.error(.invalidResponseFormat(messageType: .GQL_CONNECTION_ERROR))
+                guard let connectionAckResponse = try? decoder.decode(ConnectionAckResponse.self, from: json) else {
+                    try await error(.invalidResponseFormat(messageType: .GQL_CONNECTION_ERROR))
                     return
                 }
-                try await self.onConnectionAck(connectionAckResponse, self)
+                try await onConnectionAck(connectionAckResponse, self)
             case .GQL_CONNECTION_KEEP_ALIVE:
-                guard let connectionKeepAliveResponse = try? self.decoder.decode(ConnectionKeepAliveResponse.self, from: json) else {
-                    try await self.error(.invalidResponseFormat(messageType: .GQL_CONNECTION_KEEP_ALIVE))
+                guard let connectionKeepAliveResponse = try? decoder.decode(ConnectionKeepAliveResponse.self, from: json) else {
+                    try await error(.invalidResponseFormat(messageType: .GQL_CONNECTION_KEEP_ALIVE))
                     return
                 }
-                try await self.onConnectionKeepAlive(connectionKeepAliveResponse, self)
+                try await onConnectionKeepAlive(connectionKeepAliveResponse, self)
             case .GQL_DATA:
-                guard let nextResponse = try? self.decoder.decode(DataResponse.self, from: json) else {
-                    try await self.error(.invalidResponseFormat(messageType: .GQL_DATA))
+                guard let nextResponse = try? decoder.decode(DataResponse.self, from: json) else {
+                    try await error(.invalidResponseFormat(messageType: .GQL_DATA))
                     return
                 }
-                try await self.onData(nextResponse, self)
+                try await onData(nextResponse, self)
             case .GQL_ERROR:
-                guard let errorResponse = try? self.decoder.decode(ErrorResponse.self, from: json) else {
-                    try await self.error(.invalidResponseFormat(messageType: .GQL_ERROR))
+                guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: json) else {
+                    try await error(.invalidResponseFormat(messageType: .GQL_ERROR))
                     return
                 }
-                try await self.onError(errorResponse, self)
+                try await onError(errorResponse, self)
             case .GQL_COMPLETE:
-                guard let completeResponse = try? self.decoder.decode(CompleteResponse.self, from: json) else {
-                    try await self.error(.invalidResponseFormat(messageType: .GQL_COMPLETE))
+                guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: json) else {
+                    try await error(.invalidResponseFormat(messageType: .GQL_COMPLETE))
                     return
                 }
-                try await self.onComplete(completeResponse, self)
+                try await onComplete(completeResponse, self)
             default:
-                try await self.error(.invalidType())
+                try await error(.invalidType())
             }
         }
     }
 
-    /// Define the callback run on receipt of a `connection_error` message
-    /// - Parameter callback: The callback to assign
-    public func onConnectionError(_ callback: @escaping (ConnectionErrorResponse, Client) async throws -> Void) {
-        onConnectionError = callback
-    }
-
-    /// Define the callback run on receipt of a `connection_ack` message
-    /// - Parameter callback: The callback to assign
-    public func onConnectionAck(_ callback: @escaping (ConnectionAckResponse, Client) async throws -> Void) {
-        onConnectionAck = callback
-    }
-
-    /// Define the callback run on receipt of a `connection_ka` message
-    /// - Parameter callback: The callback to assign
-    public func onConnectionKeepAlive(_ callback: @escaping (ConnectionKeepAliveResponse, Client) async throws -> Void) {
-        onConnectionKeepAlive = callback
-    }
-
-    /// Define the callback run on receipt of a `data` message
-    /// - Parameter callback: The callback to assign
-    public func onData(_ callback: @escaping (DataResponse, Client) async throws -> Void) {
-        onData = callback
-    }
-
-    /// Define the callback run on receipt of an `error` message
-    /// - Parameter callback: The callback to assign
-    public func onError(_ callback: @escaping (ErrorResponse, Client) async throws -> Void) {
-        onError = callback
-    }
-
-    /// Define the callback run on receipt of any message
-    /// - Parameter callback: The callback to assign
-    public func onComplete(_ callback: @escaping (CompleteResponse, Client) async throws -> Void) {
-        onComplete = callback
-    }
-
-    /// Define the callback run on receipt of a `complete` message
-    /// - Parameter callback: The callback to assign
-    public func onMessage(_ callback: @escaping (String, Client) async throws -> Void) {
-        onMessage = callback
-    }
-
     /// Send a `connection_init` request through the messenger
     public func sendConnectionInit(payload: InitPayload) async throws {
-        guard let messenger = messenger else { return }
         try await messenger.send(
             ConnectionInitRequest(
                 payload: payload
@@ -144,7 +120,6 @@ public class Client<InitPayload: Equatable & Codable> {
 
     /// Send a `start` request through the messenger
     public func sendStart(payload: GraphQLRequest, id: String) async throws {
-        guard let messenger = messenger else { return }
         try await messenger.send(
             StartRequest(
                 payload: payload,
@@ -155,7 +130,6 @@ public class Client<InitPayload: Equatable & Codable> {
 
     /// Send a `stop` request through the messenger
     public func sendStop(id: String) async throws {
-        guard let messenger = messenger else { return }
         try await messenger.send(
             StopRequest(
                 id: id
@@ -165,15 +139,13 @@ public class Client<InitPayload: Equatable & Codable> {
 
     /// Send a `connection_terminate` request through the messenger
     public func sendConnectionTerminate() async throws {
-        guard let messenger = messenger else { return }
         try await messenger.send(
             ConnectionTerminateRequest().toJSON(encoder)
         )
     }
 
     /// Send an error through the messenger and close the connection
     private func error(_ error: GraphQLWSError) async throws {
-        guard let messenger = messenger else { return }
         try await messenger.error(error.message, code: error.code.rawValue)
     }
 }

Sources/GraphQLWS/GraphQLWSError.swift

@@ -89,7 +89,7 @@ struct GraphQLWSError: Error {
 
 /// Error codes for miscellaneous issues
 public enum ErrorCode: Int, CustomStringConvertible, Sendable {
-    // Miscellaneous
+    /// Miscellaneous
     case miscellaneous = 4400
 
     // Internal errors

Sources/GraphQLWS/Messenger.swift

@@ -1,15 +1,10 @@
 import Foundation
 
-/// Protocol for an object that can send and recieve messages. This allows mocking in tests
-public protocol Messenger: AnyObject {
-    // AnyObject compliance requires that the implementing object is a class and we can reference it weakly
-
+/// Protocol for an object that can send messages.
+public protocol Messenger: Sendable {
     /// Send a message through this messenger
     /// - Parameter message: The message to send
-    func send<S>(_ message: S) async throws -> Void where S: Collection, S.Element == Character
-
-    /// Set the callback that should be run when a message is recieved
-    func onReceive(callback: @escaping (String) async throws -> Void)
+    func send<S: Sendable & Collection>(_ message: S) async throws -> Void where S.Element == Character
 
     /// Close the messenger
     func close() async throws