Server/Client now wrap messenger

Client gets better message sending methods

Author: NeedleInAJayStack

Sources/GraphQLWS/Client.swift

@@ -1,10 +1,13 @@
 // Copyright (c) 2021 PassiveLogic, Inc.
 
 import Foundation
+import GraphQL
 
 /// Adds client-side [graphql-ws protocol](https://github.com/apollographql/subscriptions-transport-ws/blob/master/PROTOCOL.md)
 /// support, namely parsing and adding callbacks for each type of server respose.
 class Client {
+    let messenger: Messenger
+    
     let onMessage: (String) -> Void
     let onConnectionError: (ConnectionErrorResponse) -> Void
     let onConnectionAck: (ConnectionAckResponse) -> Void
@@ -13,37 +16,41 @@ class Client {
     let onError: (ErrorResponse) -> Void
     let onComplete: (CompleteResponse) -> Void
 
+    let encoder = GraphQLJSONEncoder()
     let decoder = JSONDecoder()
 
     /// Create a new client.
     ///
     /// - Parameters:
-    ///   - onMessage: callback run on receipt of any message
-    ///   - onConnectionError: callback run on receipt of a `connection_error` message
-    ///   - onConnectionAck: callback run on receipt of a `connection_ack` message
-    ///   - onData: callback run on receipt of a `data` message
-    ///   - onError: callback run on receipt of an `error` message
-    ///   - onComplete: callback run on receipt of a `complete` message
+    ///   - messenger: The messenger to bind the client to.
+    ///   - onConnectionError: Callback run on receipt of a `connection_error` message
+    ///   - onConnectionAck: Callback run on receipt of a `connection_ack` message
+    ///   - onData: Callback run on receipt of a `data` message
+    ///   - onError: Callback run on receipt of an `error` message
+    ///   - onComplete: Callback run on receipt of a `complete` message
+    ///   - onMessage: Callback run on receipt of any message
     init(
-        onMessage: @escaping (String) -> Void = { _ in () },
+        messenger: Messenger,
         onConnectionError: @escaping (ConnectionErrorResponse) -> Void = { _ in () },
         onConnectionAck: @escaping (ConnectionAckResponse) -> Void = { _ in () },
         onConnectionKeepAlive: @escaping (ConnectionKeepAliveResponse) -> Void = { _ in () },
         onData: @escaping (DataResponse) -> Void = { _ in () },
         onError: @escaping (ErrorResponse) -> Void = { _ in () },
-        onComplete: @escaping (CompleteResponse) -> Void = { _ in () }
+        onComplete: @escaping (CompleteResponse) -> Void = { _ in () },
+        onMessage: @escaping (String) -> Void = { _ in () }
     ) {
+        self.messenger = messenger
         self.onMessage = onMessage
         self.onConnectionError = onConnectionError
         self.onConnectionAck = onConnectionAck
         self.onConnectionKeepAlive = onConnectionKeepAlive
         self.onData = onData
         self.onError = onError
         self.onComplete = onComplete
-    }
-    
-    func attach(to messenger: Messenger) {
-        messenger.onRecieve { message in
+        
+        self.messenger.onRecieve { [weak self] message in
+            guard let self = self else { return }
+            
             self.onMessage(message)
 
             // Detect and ignore error responses.
@@ -54,7 +61,7 @@ class Client {
 
             guard let json = message.data(using: .utf8) else {
                 let error = GraphQLWSError.invalidEncoding()
-                messenger.error(error.message, code: error.code)
+                self.messenger.error(error.message, code: error.code)
                 return
             }
 
@@ -64,57 +71,92 @@ class Client {
             }
             catch {
                 let error = GraphQLWSError.noType()
-                messenger.error(error.message, code: error.code)
+                self.messenger.error(error.message, code: error.code)
                 return
             }
 
             switch response.type {
                 case .GQL_CONNECTION_ERROR:
                     guard let connectionErrorResponse = try? self.decoder.decode(ConnectionErrorResponse.self, from: json) else {
                         let error = GraphQLWSError.invalidResponseFormat(messageType: .GQL_CONNECTION_ERROR)
-                        messenger.error(error.message, code: error.code)
+                        self.messenger.error(error.message, code: error.code)
                         return
                     }
                     self.onConnectionError(connectionErrorResponse)
                 case .GQL_CONNECTION_ACK:
                     guard let connectionAckResponse = try? self.decoder.decode(ConnectionAckResponse.self, from: json) else {
                         let error = GraphQLWSError.invalidResponseFormat(messageType: .GQL_CONNECTION_ACK)
-                        messenger.error(error.message, code: error.code)
+                        self.messenger.error(error.message, code: error.code)
                         return
                     }
                     self.onConnectionAck(connectionAckResponse)
                 case .GQL_CONNECTION_KEEP_ALIVE:
                     guard let connectionKeepAliveResponse = try? self.decoder.decode(ConnectionKeepAliveResponse.self, from: json) else {
                         let error = GraphQLWSError.invalidResponseFormat(messageType: .GQL_CONNECTION_KEEP_ALIVE)
-                        messenger.error(error.message, code: error.code)
+                        self.messenger.error(error.message, code: error.code)
                         return
                     }
                     self.onConnectionKeepAlive(connectionKeepAliveResponse)
                 case .GQL_DATA:
                     guard let nextResponse = try? self.decoder.decode(DataResponse.self, from: json) else {
                         let error = GraphQLWSError.invalidResponseFormat(messageType: .GQL_DATA)
-                        messenger.error(error.message, code: error.code)
+                        self.messenger.error(error.message, code: error.code)
                         return
                     }
                     self.onData(nextResponse)
                 case .GQL_ERROR:
                     guard let errorResponse = try? self.decoder.decode(ErrorResponse.self, from: json) else {
                         let error = GraphQLWSError.invalidResponseFormat(messageType: .GQL_ERROR)
-                        messenger.error(error.message, code: error.code)
+                        self.messenger.error(error.message, code: error.code)
                         return
                     }
                     self.onError(errorResponse)
                 case .GQL_COMPLETE:
                     guard let completeResponse = try? self.decoder.decode(CompleteResponse.self, from: json) else {
                         let error = GraphQLWSError.invalidResponseFormat(messageType: .GQL_COMPLETE)
-                        messenger.error(error.message, code: error.code)
+                        self.messenger.error(error.message, code: error.code)
                         return
                     }
                     self.onComplete(completeResponse)
                 case .unknown:
                     let error = GraphQLWSError.invalidType()
-                    messenger.error(error.message, code: error.code)
+                    self.messenger.error(error.message, code: error.code)
             }
         }
     }
+    
+    /// Send a `connection_init` request through the messenger
+    func sendConnectionInit(payload: ConnectionInitAuth?) {
+        messenger.send(
+            ConnectionInitRequest(
+                payload: payload
+            ).toJSON(encoder)
+        )
+    }
+    
+    /// Send a `start` request through the messenger
+    func sendStart(payload: GraphQLRequest, id: String) {
+        messenger.send(
+            StartRequest(
+                payload: payload,
+                id: id
+            ).toJSON(encoder)
+        )
+    }
+    
+    /// Send a `stop` request through the messenger
+    func sendStop(id: String) {
+        messenger.send(
+            StopRequest(
+                id: id
+            ).toJSON(encoder)
+        )
+    }
+    
+    /// Send a `connection_terminate` request through the messenger
+    func sendConnectionTerminate() {
+        messenger.send(
+            ConnectionTerminateRequest().toJSON(encoder)
+        )
+    }
 }

Sources/GraphQLWS/Server.swift

@@ -7,8 +7,10 @@ import NIO
 import RxSwift
 
 /// Adds server-side [graphql-ws subprotocol](https://github.com/apollographql/subscriptions-transport-ws/blob/master/PROTOCOL.md)
-/// support, namely parsing and adding callbacks for each type of client request.
+/// support. This handles the majority of query processing according to the procol definition, allowing a few callbacks for customization.
 class Server {
+    let messenger: Messenger
+    
     let auth: (ConnectionInitRequest) throws -> Void
     let onExecute: (GraphQLRequest) -> EventLoopFuture<GraphQLResult>
     let onSubscribe: (GraphQLRequest) -> EventLoopFuture<SubscriptionResult>
@@ -21,24 +23,33 @@ class Server {
     let decoder = JSONDecoder()
     let encoder = GraphQLJSONEncoder()
 
+    /// Create a new server
+    ///
+    /// - Parameters:
+    ///   - messenger: The messenger to bind the server to.
+    ///   - auth: Callback run during `connection_init` resolution that allows authorization using the `payload`. Throw to indicate that authorization has failed.
+    ///   - onExecute: Callback run during `start` resolution for non-streaming queries. Typically this is `API.execute`.
+    ///   - onSubscribe: Callback run during `start` resolution for streaming queries. Typically this is `API.subscribe`.
+    ///   - onExit: Callback run when the communication is shut down, either by the client or server
+    ///   - onMessage: callback run on receipt of any message
     init(
+        messenger: Messenger,
         auth: @escaping (ConnectionInitRequest) throws -> Void,
         onExecute: @escaping (GraphQLRequest) -> EventLoopFuture<GraphQLResult>,
         onSubscribe: @escaping (GraphQLRequest) -> EventLoopFuture<SubscriptionResult>,
         onExit: @escaping () -> Void,
         onMessage: @escaping (String) -> Void = { _ in () }
     ) {
+        self.messenger = messenger
         self.auth = auth
         self.onExecute = onExecute
         self.onSubscribe = onSubscribe
         self.onExit = onExit
         self.onMessage = onMessage
-    }
-    
-    /// Attaches the responder to the provided Messenger in order to recieve and transmit messages
-    /// - Parameter messenger: The Messenger to use for communication
-    func attach(to messenger: Messenger) {
-        messenger.onRecieve { message in
+        
+        self.messenger.onRecieve { [weak self] message in
+            guard let self = self else { return }
+            
             self.onMessage(message)
 
             // Detect and ignore error responses.
@@ -49,7 +60,7 @@ class Server {
 
             guard let json = message.data(using: .utf8) else {
                 let error = GraphQLWSError.invalidEncoding()
-                messenger.error(error.message, code: error.code)
+                self.messenger.error(error.message, code: error.code)
                 return
             }
 
@@ -59,53 +70,53 @@ class Server {
             }
             catch {
                 let error = GraphQLWSError.noType()
-                messenger.error(error.message, code: error.code)
+                self.messenger.error(error.message, code: error.code)
                 return
             }
 
             switch request.type {
                 case .GQL_CONNECTION_INIT:
                     guard let connectionInitRequest = try? self.decoder.decode(ConnectionInitRequest.self, from: json) else {
                         let error = GraphQLWSError.invalidRequestFormat(messageType: .GQL_CONNECTION_INIT)
-                        messenger.error(error.message, code: error.code)
+                        self.messenger.error(error.message, code: error.code)
                         return
                     }
-                    self.onConnectionInit(connectionInitRequest, messenger)
+                    self.onConnectionInit(connectionInitRequest)
                 case .GQL_START:
                     guard let startRequest = try? self.decoder.decode(StartRequest.self, from: json) else {
                         let error = GraphQLWSError.invalidRequestFormat(messageType: .GQL_START)
-                        messenger.error(error.message, code: error.code)
+                        self.messenger.error(error.message, code: error.code)
                         return
                     }
-                    self.onStart(startRequest, messenger)
+                    self.onStart(startRequest)
                 case .GQL_STOP:
                     guard let stopRequest = try? self.decoder.decode(StopRequest.self, from: json) else {
                         let error = GraphQLWSError.invalidRequestFormat(messageType: .GQL_STOP)
-                        messenger.error(error.message, code: error.code)
+                        self.messenger.error(error.message, code: error.code)
                         return
                     }
-                    self.onStop(stopRequest, messenger)
+                    self.onStop(stopRequest, self.messenger)
                 case .GQL_CONNECTION_TERMINATE:
                     guard let connectionTerminateRequest = try? self.decoder.decode(ConnectionTerminateRequest.self, from: json) else {
                         let error = GraphQLWSError.invalidRequestFormat(messageType: .GQL_CONNECTION_TERMINATE)
-                        messenger.error(error.message, code: error.code)
+                        self.messenger.error(error.message, code: error.code)
                         return
                     }
-                    self.onConnectionTerminate(connectionTerminateRequest, messenger)
+                    self.onConnectionTerminate(connectionTerminateRequest)
                 case .unknown:
                     let error = GraphQLWSError.invalidType()
-                    messenger.error(error.message, code: error.code)
+                    self.messenger.error(error.message, code: error.code)
             }
         }
 
         // Clean up any uncompleted subscriptions
         // TODO: Re-enable this
-//        messenger.onClose {
-//            _ = self.context?.cleanupSubscription()
-//        }
+        //        messenger.onClose {
+        //            _ = self.context?.cleanupSubscription()
+        //        }
     }
 
-    private func onConnectionInit(_ connectionInitRequest: ConnectionInitRequest, _ messenger: Messenger) {
+    private func onConnectionInit(_ connectionInitRequest: ConnectionInitRequest) {
         guard !initialized else {
             let error = GraphQLWSError.tooManyInitializations()
             messenger.error(error.message, code: error.code)
@@ -127,7 +138,7 @@ class Server {
         // TODO: Should we send the `ka` message?
     }
 
-    private func onStart(_ startRequest: StartRequest, _ messenger: Messenger) {
+    private func onStart(_ startRequest: StartRequest) {
         guard initialized else {
             let error = GraphQLWSError.notInitialized()
             messenger.error(error.message, code: error.code)
@@ -153,43 +164,43 @@ class Server {
                 guard let streamOpt = result.stream else {
                     // API issue - subscribe resolver isn't stream
                     let error = GraphQLWSError.internalAPIStreamIssue()
-                    messenger.error(error.message, code: error.code)
+                    self.messenger.error(error.message, code: error.code)
                     return
                 }
                 let stream = streamOpt as! ObservableSubscriptionEventStream
                 let observable = stream.observable
                 observable.subscribe(
                     onNext: { resultFuture in
                         resultFuture.whenSuccess { result in
-                            messenger.send(DataResponse(result, id: id).toJSON(self.encoder))
+                            self.messenger.send(DataResponse(result, id: id).toJSON(self.encoder))
                         }
                         resultFuture.whenFailure { error in
-                            messenger.send(ErrorResponse(error, id: id).toJSON(self.encoder))
+                            self.messenger.send(ErrorResponse(error, id: id).toJSON(self.encoder))
                         }
                     },
                     onError: { error in
-                        messenger.send(ErrorResponse(error, id: id).toJSON(self.encoder))
+                        self.messenger.send(ErrorResponse(error, id: id).toJSON(self.encoder))
                     },
                     onCompleted: {
-                        messenger.send(CompleteResponse(id: id).toJSON(self.encoder))
-                        _ = messenger.close()
+                        self.messenger.send(CompleteResponse(id: id).toJSON(self.encoder))
+                        _ = self.messenger.close()
                     }
                 ).disposed(by: self.disposeBag)
             }
             subscribeFuture.whenFailure { error in
                 let error = GraphQLWSError.graphQLError(error)
-                _ = messenger.error(error.message, code: error.code)
+                _ = self.messenger.error(error.message, code: error.code)
             }
         }
         else {
             let executeFuture = onExecute(graphQLRequest)
             executeFuture.whenSuccess { result in
-                messenger.send(DataResponse(result, id: id).toJSON(self.encoder))
-                messenger.send(CompleteResponse(id: id).toJSON(self.encoder))
+                self.messenger.send(DataResponse(result, id: id).toJSON(self.encoder))
+                self.messenger.send(CompleteResponse(id: id).toJSON(self.encoder))
             }
             executeFuture.whenFailure { error in
-                messenger.send(ErrorResponse(error, id: id).toJSON(self.encoder))
-                messenger.send(CompleteResponse(id: id).toJSON(self.encoder))
+                self.messenger.send(ErrorResponse(error, id: id).toJSON(self.encoder))
+                self.messenger.send(CompleteResponse(id: id).toJSON(self.encoder))
             }
         }
     }
@@ -203,7 +214,7 @@ class Server {
         onExit()
     }
 
-    private func onConnectionTerminate(_: ConnectionTerminateRequest, _ messenger: Messenger) {
+    private func onConnectionTerminate(_: ConnectionTerminateRequest) {
         onExit()
         _ = messenger.close()
     }