Merge pull request #1 from PassiveLogic/feature/two-way-protocol

Allow `Next` messages from Client -> Server

Author: Garrett Moseke

.swiftformat

@@ -0,0 +1,94 @@
+# Default formatting rules for all PassiveLogic swift projects
+
+--allman false
+--binarygrouping none
+--decimalgrouping none
+--disable redundantSelf,trailingClosures
+--elseposition next-line
+--exponentcase uppercase
+--guardelse same-line
+--hexliteralcase lowercase
+--indent 4
+--indentcase true
+--maxwidth 140
+--self insert
+--semicolons never
+--swiftversion 5
+--trimwhitespace always
+--varattributes prev-line
+--typeattributes prev-line
+--wrapparameters before-first
+--wrapcollections before-first
+--wraparguments before-first
+# By default we ignore anything in the "Packages" directory as this includes Third Party code.
+--exclude Packages
+
+# Explicitly enable rules to prevent automatic opt-in
+# Generated from `swiftformat --rules` on v0.47.1
+--rules andOperator
+--rules anyObjectProtocol
+--rules blankLinesAroundMark
+--rules blankLinesAtEndOfScope
+--rules blankLinesAtStartOfScope
+--rules blankLinesBetweenScopes
+--rules braces
+--rules consecutiveBlankLines
+--rules consecutiveSpaces
+--rules duplicateImports
+--rules elseOnSameLine
+--rules emptyBraces
+--rules enumNamespaces
+--rules extensionAccessControl
+--rules fileHeader
+--rules hoistPatternLet
+--rules indent
+--rules initCoderUnavailable
+--rules leadingDelimiters
+--rules linebreakAtEndOfFile
+--rules linebreaks
+--rules modifierOrder
+--rules numberFormatting
+--rules preferKeyPath
+--rules redundantBackticks
+--rules redundantBreak
+--rules redundantExtensionACL
+--rules redundantFileprivate
+--rules redundantGet
+--rules redundantInit
+--rules redundantLet
+--rules redundantLetError
+--rules redundantNilInit
+--rules redundantObjc
+--rules redundantParens
+--rules redundantPattern
+--rules redundantRawValues
+--rules redundantReturn
+--rules redundantSelf
+--rules redundantType
+--disable redundantVoidReturnType  # Disabled because this breaks some NIO closures
+--rules semicolons
+--rules sortedImports
+--rules spaceAroundBraces
+--rules spaceAroundBrackets
+--rules spaceAroundComments
+--rules spaceAroundGenerics
+--rules spaceAroundOperators
+--rules spaceAroundParens
+--rules spaceInsideBraces
+--rules spaceInsideBrackets
+--rules spaceInsideComments
+--rules spaceInsideGenerics
+--rules spaceInsideParens
+--rules strongOutlets
+--rules strongifiedSelf
+--rules todos
+--rules trailingClosures
+--rules trailingCommas
+--rules trailingSpace
+--rules unusedArguments
+--rules void
+--rules wrap
+--rules wrapArguments
+--rules wrapAttributes
+--rules wrapMultilineStatementBraces
+--rules yodaConditions

Package.swift

@@ -25,8 +25,9 @@ let package = Package(
                 .product(name: "GraphQLRxSwift", package: "GraphQLRxSwift"),
                 .product(name: "GraphQL", package: "GraphQL"),
                 .product(name: "NIO", package: "swift-nio"),
-                .product(name: "RxSwift", package: "RxSwift")
-            ]),
+                .product(name: "RxSwift", package: "RxSwift"),
+            ]
+        ),
         .testTarget(
             name: "GraphQLTransportWSTests",
             dependencies: ["GraphQLTransportWS"]

README.md

@@ -1,14 +1,36 @@
-# GraphQLTransportWS
+# GraphQLTransportWS-DataSync
 
-This implements the [graphql-transport-ws WebSocket subprotocol](https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md).
-It is mainly intended for server support, but there is a basic client implementation included.
+This implements the [graphql-transport-ws WebSocket subprotocol](https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md), with the additional capability to send `GraphQLRequests` from the Client to the Server over the same websocket to support the PassiveLogic DataSync spec.
+
+It is mainly intended for server support, but there is a client implementation included for swift projects.
 
 Features:
 - Server implementation that implements defined protocol conversations
 - Client and Server types that wrap messengers
 - Codable Server and Client message structures
 - Custom authentication support
 
+## DataSync Additions
+This DataSync fork allows for `Next` messages to be handled by the Server with a custom `onNext` callback exposed. The server will detect incoming `Next` frames that contain `subscribe` requests and reject them with an error to avoid nesting subscriptions, but will allow handling logic for `query` and `mutation` requests.
+
+The Client implemeneted here now also supports sending `Next` messages to the server via adding an `Observable` object, which will automatically wrap and send `Next` frames as it updates.
+
+### Example
+*The client and the server has already gone through [successful connection initialisation.](https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md#successful-connection-initialisation)*
+1. *Client* generates a unique ID for the following operation
+2. *Client* dispatches the `Subscribe` message with the generated ID through the `id` field and the requested operation passed through the `payload` field
+*All future communication is linked through this unique ID*
+3. *Server* executes the streaming GraphQL operation
+4. *Server* checks if the generated ID is unique across active streaming subscriptions
+    - If not unique, the server will close the socket with the event `4409: Subscriber for <generated-id> already exists`
+    If unique, continue...
+5. *Server* optionally checks if the operation is valid before starting executing it, e.g. checking permissions
+    - If not valid, the server sends an `Error` message and deems the operation complete.
+    If valid, continue...
+6. *Server* & *Client* dispatch results over time with the `Next` message
+    - *Server* & *Client* handle received `Next` messages with their respectively defined callbacks. All Context is assumed to be the same between the two and is established in step 3 when the subscription is created.
+7. The operation completes as described in [the `graphql-transport-ws` protocol](https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md#streaming-operation)
+
 ## Usage
 
 To use this package, include it in your `Package.swift` dependencies:
@@ -89,6 +111,28 @@ routes.webSocket(
 )
 ```
 
+You can also Create a client and add an observable callback to send data back to the server:
+```swift
+let messenger = WebSocketMessenger(websocket: websocket)
+let client = Client<EmptyInitPayload>(messenger: messenger)
+
+// add an observable to the client that will automatically send `Next` frames to the server as it updates
+client.addObservable(observable: 
+    Observable.create { observer in
+    // Some simple obsererver that fulfills every future with the same query. Your observer can be much more complex.
+    observer.on(.next(loop.makeSucceededFuture(GraphQLRequest(
+            query:
+                """
+                query {
+                    hello
+                }
+                """
+        ))))
+        return Disposables.create()
+    }
+)
+```
+
 ### Authentication
 
 This package exposes authentication hooks on the `connection_init` message. To perform custom authentication,

Sources/GraphQLTransportWS/Client.swift

@@ -1,22 +1,24 @@
-// Copyright (c) 2021 PassiveLogic, Inc.
-
 import Foundation
 import GraphQL
+import GraphQLRxSwift
+import NIO
+import RxSwift
 
 /// 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?
-    
+
     var onConnectionAck: (ConnectionAckResponse, Client) -> Void = { _, _ in }
     var onNext: (NextResponse, Client) -> Void = { _, _ in }
     var onError: (ErrorResponse, Client) -> Void = { _, _ in }
     var onComplete: (CompleteResponse, Client) -> Void = { _, _ in }
     var onMessage: (String, Client) -> Void = { _, _ in }
-    
+
+    let disposeBag = DisposeBag()
     let encoder = GraphQLJSONEncoder()
     let decoder = JSONDecoder()
-    
+
     /// Create a new client.
     ///
     /// - Parameters:
@@ -27,18 +29,18 @@ public class Client<InitPayload: Equatable & Codable> {
         self.messenger = messenger
         messenger.onReceive { message in
             self.onMessage(message, self)
-            
+
             // 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 {
+
+            guard let json = Data(message.utf8) else {
                 self.error(.invalidEncoding())
                 return
             }
-            
+
             let response: Response
             do {
                 response = try self.decoder.decode(Response.self, from: json)
@@ -47,7 +49,7 @@ public class Client<InitPayload: Equatable & Codable> {
                 self.error(.noType())
                 return
             }
-            
+
             switch response.type {
                 case .connectionAck:
                     guard let connectionAckResponse = try? self.decoder.decode(ConnectionAckResponse.self, from: json) else {
@@ -78,37 +80,37 @@ public class Client<InitPayload: Equatable & Codable> {
             }
         }
     }
-    
+
     /// Define the callback run on receipt of a `connection_ack` message
     /// - Parameter callback: The callback to assign
     public func onConnectionAck(_ callback: @escaping (ConnectionAckResponse, Client) -> Void) {
         self.onConnectionAck = callback
     }
-    
+
     /// Define the callback run on receipt of a `next` message
     /// - Parameter callback: The callback to assign
     public func onNext(_ callback: @escaping (NextResponse, Client) -> Void) {
         self.onNext = callback
     }
-    
+
     /// Define the callback run on receipt of an `error` message
     /// - Parameter callback: The callback to assign
     public func onError(_ callback: @escaping (ErrorResponse, Client) -> Void) {
         self.onError = callback
     }
-    
+
     /// Define the callback run on receipt of a `complete` message
     /// - Parameter callback: The callback to assign
     public func onComplete(_ callback: @escaping (CompleteResponse, Client) -> Void) {
         self.onComplete = callback
     }
-    
+
     /// Define the callback run on receipt of any message
     /// - Parameter callback: The callback to assign
     public func onMessage(_ callback: @escaping (String, Client) -> Void) {
         self.onMessage = callback
     }
-    
+
     /// Send a `connection_init` request through the messenger
     public func sendConnectionInit(payload: InitPayload) {
         guard let messenger = messenger else { return }
@@ -118,9 +120,9 @@ public class Client<InitPayload: Equatable & Codable> {
             ).toJSON(encoder)
         )
     }
-    
+
     /// Send a `subscribe` request through the messenger
-    public func sendStart(payload: GraphQLRequest, id: String) {
+    public func sendSubscribe(payload: GraphQLRequest, id: String) {
         guard let messenger = messenger else { return }
         messenger.send(
             SubscribeRequest(
@@ -129,7 +131,7 @@ public class Client<InitPayload: Equatable & Codable> {
             ).toJSON(encoder)
         )
     }
-    
+
     /// Send a `complete` request through the messenger
     public func sendStop(id: String) {
         guard let messenger = messenger else { return }
@@ -139,7 +141,37 @@ public class Client<InitPayload: Equatable & Codable> {
             ).toJSON(encoder)
         )
     }
-    
+
+    /// Add an observable object for this client that will fire off `Next` messages to the server as updates happen.
+    /// - Parameter observable: `Observable<EventLoopFuture<GraphQLRequest>>` to subscribe to for changes.
+    public func addObservableSubscription(observable: Observable<EventLoopFuture<GraphQLResult>>) {
+        observable.subscribe(
+            onNext: { [weak self] resultFuture in
+                guard let self = self else { return }
+                resultFuture.whenSuccess { result in
+                    self.sendNext(payload: result, id: UUID().uuidString)
+                }
+                resultFuture.whenFailure { error in
+                    self.error(.graphQLError(error))
+                }
+            }
+        ).disposed(by: disposeBag)
+    }
+
+    /// Send a `next` message through the messenger.
+    /// - Parameters:
+    ///   - payload: `GraphQLRequest` object for the server to handle
+    ///   - id: id of the message
+    private func sendNext(payload: GraphQLResult, id: String) {
+        guard let messenger = messenger else { return }
+        messenger.send(
+            NextResponse(
+                payload,
+                id: id
+            ).toJSON(encoder)
+        )
+    }
+
     /// Send an error through the messenger and close the connection
     private func error(_ error: GraphQLTransportWSError) {
         guard let messenger = messenger else { return }