WebSocket Transport (#43)

Author: dnadoba

Package.swift

@@ -17,7 +17,7 @@ let package = Package(
         .library(name: "RSocketNIOChannel", targets: ["RSocketNIOChannel"]),
 
         // Transport protocol
-        .library(name: "RSocketWebSocketTransport", targets: ["RSocketWebSocketTransport"]),
+        .library(name: "RSocketWSTransport", targets: ["RSocketWSTransport"]),
         .library(name: "RSocketTCPTransport", targets: ["RSocketTCPTransport"]),
 
         // Examples
@@ -60,7 +60,7 @@ let package = Package(
         ]),
 
         // Transport protocol
-        .target(name: "RSocketWebSocketTransport", dependencies: [
+        .target(name: "RSocketWSTransport", dependencies: [
             "RSocketCore",
             .product(name: "NIO", package: "swift-nio"),
             .product(name: "NIOHTTP1", package: "swift-nio"),
@@ -94,14 +94,17 @@ let package = Package(
             "ReactiveSwift",
             .product(name: "NIO", package: "swift-nio"),
         ]),
+        .testTarget(name: "RSocketWSTransportTests", dependencies: [
+            "RSocketWSTransport"
+        ]),
 
         // Examples
         .target(
             name: "TimerClientExample",
             dependencies: [
                 "RSocketCore",
                 "RSocketNIOChannel",
-                "RSocketWebSocketTransport",
+                "RSocketWSTransport",
                 "RSocketReactiveSwift",
                 .product(name: "ReactiveSwift", package: "ReactiveSwift"),
                 .product(name: "ArgumentParser", package: "swift-argument-parser"),
@@ -113,7 +116,7 @@ let package = Package(
             dependencies: [
                 "RSocketCore",
                 "RSocketNIOChannel",
-                "RSocketWebSocketTransport",
+                "RSocketWSTransport",
                 "RSocketReactiveSwift",
                 .product(name: "ReactiveSwift", package: "ReactiveSwift"),
                 .product(name: "ArgumentParser", package: "swift-argument-parser"),

Sources/Examples/TimerClient/main.swift

@@ -4,7 +4,7 @@ import ReactiveSwift
 import RSocketCore
 import RSocketNIOChannel
 import RSocketReactiveSwift
-import RSocketWebSocketTransport
+import RSocketWSTransport
 
 func route(_ route: String) -> Data {
     let encodedRoute = Data(route.utf8)
@@ -14,56 +14,48 @@ func route(_ route: String) -> Data {
     return encodedRouteLength + encodedRoute
 }
 
+extension URL: ExpressibleByArgument {
+    public init?(argument: String) {
+        guard let url = URL(string: argument) else { return nil }
+        self = url
+    }
+    public var defaultValueDescription: String { description }
+}
+
 /// the server-side code can be found here -> https://github.com/rsocket/rsocket-demo/tree/master/src/main/kotlin/io/rsocket/demo/timer
 struct TimerClientExample: ParsableCommand {
     static var configuration = CommandConfiguration(
         abstract: "connects to an RSocket endpoint using WebSocket transport, requests a stream at the route `timer` and logs all events."
     )
 
     @Option
-    var host = "demo.rsocket.io/rsocket"
-    
-    @Option
-    var port = 80
+    var url = URL(string: "wss://demo.rsocket.io/rsocket")!
 
     @Option(help: "maximum number of responses that are taken before it cancels the stream")
     var limit = 10000
 
-    mutating func run() throws {
+    func run() throws {
         let bootstrap = ClientBootstrap(
-                config: ClientSetupConfig(
-                        timeBetweenKeepaliveFrames: 0,
-                        maxLifetime: 30_000,
-                        metadataEncodingMimeType: "message/x.rsocket.routing.v0",
-                        dataEncodingMimeType: "application/json"
-                ),
-                transport: WSTransport(),
-                timeout: .seconds(30)
+            config: ClientSetupConfig(
+                timeBetweenKeepaliveFrames: 30_000,
+                maxLifetime: 60_000,
+                metadataEncodingMimeType: "message/x.rsocket.routing.v0",
+                dataEncodingMimeType: "application/json"
+            ),
+            transport: WSTransport()
         )
-
-        let clientProducer = bootstrap.connect(host: host, port: port)
-
-        let clientProperty = Property<ReactiveSwiftClient?>(initial: nil, then: clientProducer.flatMapError { _ in
-            .empty
-        })
-
-        let streamSemaphore = DispatchSemaphore(value: 0)
-        clientProperty
-            .producer
-            .skipNil()
-            .flatMap(.latest) {
-                $0.requester.requestStream(payload: Payload(
-                    metadata: route("timer"),
-                    data: Data()
-                ))
-            }
-            .map() { $0.data }
-            .logEvents(identifier: "route.timer")
-            .take(first: limit)
-            .on(disposed: { streamSemaphore.signal() })
-            .start()
-
-        streamSemaphore.wait()
+        
+        let client = try bootstrap.connect(to: .init(url: url)).first()!.get()
+
+        try client.requester.requestStream(payload: Payload(
+            metadata: route("timer"),
+            data: Data()
+        ))
+        .map() { String.init(decoding: $0.data, as: UTF8.self) }
+        .logEvents(identifier: "route.timer")
+        .take(first: limit)
+        .wait()
+        .get()
     }
 }
 

Sources/Examples/TwitterClient/main.swift

@@ -4,7 +4,7 @@ import ReactiveSwift
 import RSocketCore
 import RSocketNIOChannel
 import RSocketReactiveSwift
-import RSocketWebSocketTransport
+import RSocketWSTransport
 
 func route(_ route: String) -> Data {
     let encodedRoute = Data(route.utf8)
@@ -14,58 +14,56 @@ func route(_ route: String) -> Data {
     return encodedRouteLength + encodedRoute
 }
 
+extension URL: ExpressibleByArgument {
+    public init?(argument: String) {
+        guard let url = URL(string: argument) else { return nil }
+        self = url
+    }
+    public var defaultValueDescription: String { description }
+}
+
 /// the server-side code can be found here -> https://github.com/rsocket/rsocket-demo/tree/master/src/main/kotlin/io/rsocket/demo/twitter
 struct TwitterClientExample: ParsableCommand {
     static var configuration = CommandConfiguration(
         abstract: "connects to an RSocket endpoint using WebSocket transport, requests a stream at the route `searchTweets` to search for tweets that match the `searchString` and logs all events."
     )
 
     @Argument(help: "used to find tweets that match the given string")
-    var searchString = "RSocket"
-    
-    @Option
-    var host = "demo.rsocket.io/rsocket"
+    var searchString = "spring"
 
     @Option
-    var port = 80
+    var url = URL(string: "wss://demo.rsocket.io/rsocket")!
 
     @Option(help: "maximum number of tweets that are taken before it cancels the stream")
     var limit = 1000
 
-    mutating func run() throws {
+    func run() throws {
         let bootstrap = ClientBootstrap(
-                config: ClientSetupConfig(
-                        timeBetweenKeepaliveFrames: 0,
-                        maxLifetime: 30_000,
-                        metadataEncodingMimeType: "message/x.rsocket.routing.v0",
-                        dataEncodingMimeType: "application/json"
-                ),
-                transport: WSTransport(),
-                timeout: .seconds(30)
+            config: ClientSetupConfig(
+                timeBetweenKeepaliveFrames: 30_000,
+                maxLifetime: 60_000,
+                metadataEncodingMimeType: "message/x.rsocket.routing.v0",
+                dataEncodingMimeType: "application/json"
+            ),
+            transport: WSTransport()
         )
+        
+        let client = try bootstrap.connect(to: .init(url: url)).first()!.get()
 
-        let clientProducer = bootstrap.connect(host: host, port: port)
-
-        let clientProperty = Property<ReactiveSwiftClient?>(initial: nil, then: clientProducer.flatMapError { _ in
-            .empty
-        })
-
-        let streamSemaphore = DispatchSemaphore(value: 0)
-        let searchString = self.searchString
-        clientProperty.producer
-                .skipNil()
-                .flatMap(.latest) {
-                    $0.requester.requestStream(payload: Payload(
-                            metadata: route("searchTweets"),
-                            data: Data(searchString.utf8)
-                    ))
-                }
-                .logEvents(identifier: "route.searchTweets")
-                .take(first: limit)
-                .on(disposed: { streamSemaphore.signal() })
-                .start()
-
-        streamSemaphore.wait();
+        try client.requester.requestStream(payload: Payload(
+            metadata: route("searchTweets"),
+            data: Data(searchString.utf8)
+        ))
+        .attemptMap { payload -> String in
+            // pretty print json
+            let json = try JSONSerialization.jsonObject(with: payload.data, options: [])
+            let data = try JSONSerialization.data(withJSONObject: json, options: [.prettyPrinted])
+            return String(decoding: data, as: UTF8.self)
+        }
+        .logEvents(identifier: "route.searchTweets")
+        .take(first: limit)
+        .wait()
+        .get()
     }
 }
 

Sources/Examples/VanillaClient/main.swift

@@ -17,30 +17,21 @@ struct VanillaClientExample: ParsableCommand {
     @Option
     var port = 7000
 
-    mutating func run() throws {
+    func run() throws {
         let bootstrap = ClientBootstrap(
-                config: ClientSetupConfig(
-                        timeBetweenKeepaliveFrames: 0,
-                        maxLifetime: 30_000,
-                        metadataEncodingMimeType: "application/octet-stream",
-                        dataEncodingMimeType: "application/octet-stream"
-                ),
-                transport: TCPTransport(),
-                timeout: .seconds(30)
+            config: ClientSetupConfig(
+                timeBetweenKeepaliveFrames: 30_000,
+                maxLifetime: 60_000,
+                metadataEncodingMimeType: "application/octet-stream",
+                dataEncodingMimeType: "application/octet-stream"
+            ),
+            transport: TCPTransport()
         )
+        
+        let client = try bootstrap.connect(to: .init(host: host, port: port)).first()!.get()
 
-        let clientProducer = bootstrap.connect(host: host, port: port)
-
-        let client: Property<ReactiveSwiftClient?> = Property(initial: nil, then: clientProducer.flatMapError { _ in
-            .empty
-        })
-
-        let streamProducer: SignalProducer<Payload, Swift.Error> = client.producer.skipNil().flatMap(.latest) {
-            $0.requester.requestStream(payload: .empty)
-        }
-        let requestProducer: SignalProducer<Payload, Swift.Error> = client.producer.skipNil().flatMap(.latest) {
-            $0.requester.requestResponse(payload: Payload(data: Data("HelloWorld".utf8)))
-        }
+        let streamProducer = client.requester.requestStream(payload: .empty)
+        let requestProducer = client.requester.requestResponse(payload: Payload(data: Data("HelloWorld".utf8)))
 
         streamProducer.logEvents(identifier: "stream1").take(first: 1).start()
         streamProducer.logEvents(identifier: "stream3").take(first: 10).start()

Sources/RSocketCore/Channel Handler/SetupWriter.swift

@@ -27,8 +27,18 @@ internal final class SetupWriter: ChannelInboundHandler, RemovableChannelHandler
         self.setup = config
         self.connectedPromise = connectedPromise
     }
-
+    
+    func handlerAdded(context: ChannelHandlerContext) {
+        if context.channel.isActive {
+            onActive(context: context)
+        }
+    }
+    
     func channelActive(context: ChannelHandlerContext) {
+        onActive(context: context)
+    }
+    
+    private func onActive(context: ChannelHandlerContext) {
         context.writeAndFlush(self.wrapOutboundOut(SetupFrameBody(
             honorsLease: false,
             version: .current,
@@ -39,7 +49,6 @@ internal final class SetupWriter: ChannelInboundHandler, RemovableChannelHandler
             dataEncodingMimeType: setup.dataEncodingMimeType,
             payload: setup.payload
         ).asFrame()), promise: nil)
-        context.fireChannelActive()
         context.channel.pipeline.removeHandler(context: context).eventLoop.assertInEventLoop()
         connectedPromise?.succeed(())
     }

Sources/RSocketCore/Client/ClientBootstrap.swift

@@ -19,5 +19,9 @@ import NIO
 public protocol ClientBootstrap {
     associatedtype Client
     associatedtype Responder
-    func connect(host: String, port: Int, responder: Responder?) -> EventLoopFuture<Client>
+    associatedtype Transport: TransportChannelHandler
+    func connect(
+        to endpoint: Transport.Endpoint,
+        responder: Responder?
+    ) -> EventLoopFuture<Client>
 }

Sources/RSocketCore/Client/TransportChannelHandler.swift

@@ -15,12 +15,21 @@
  */
 
 import NIO
+import Foundation
+
+public protocol Endpoint {
+    var host: String { get }
+    var port: Int { get }
+    
+    /// if true but TLS is not configured, connecting to this endpoint will fail
+    var requiresTLS: Bool { get }
+}
 
 public protocol TransportChannelHandler {
+    associatedtype Endpoint: RSocketCore.Endpoint
     func addChannelHandler(
         channel: Channel,
-        host: String,
-        port: Int,
+        endpoint: Endpoint,
         upgradeComplete: @escaping () -> EventLoopFuture<Void>
     ) -> EventLoopFuture<Void>
 }