Prepare Release 1.0.0.4-preview (#79)

* Add action for dev (#72)

* Add message buffer (#73)

* Add message buffer

* Some improvement

* Improve again

* Add client to server stream (#74)

* Add client to server stream

* increase timeout

* Add public initializer to HttpConnectionOptions to allow external instantiation (#76)

* hide stateful reconnect, prepare release for 1.0.0-preview.4 (#78)

* Hide stateful reconnect options

* Add public init for MsgpackTimestamp

* Release 1.0.0-preview.4

* Update readme

---------

Co-authored-by: Chenyang Liu <zackliu1995@hotmail.com>
Co-authored-by: Yannis Alexiou <yannisalexiou@users.noreply.github.com>

Author: 3 people

.github/workflows/swift.yml

@@ -5,9 +5,9 @@ name: Swift
 
 on:
   push:
-    branches: [ "main" ]
+    branches: [ "main", "dev" ]
   pull_request:
-    branches: [ "main" ]
+    branches: [ "main", "dev" ]
 
 jobs:
   build:

CHANGELOG.md

@@ -1,4 +1,13 @@
 # Release History
+## 1.0.0-preview.4
+
+### Bugs Fixed
+
+- Fix the accessbility of HttpConnectionOptions
+
+### Features Added
+
+- Support client to server streaming
 
 ## 1.0.0-preview.3
 
@@ -15,4 +24,4 @@
 ## 1.0.0-preview.1
 
 ### Features Added
-- The initial preview release of the library
+- The initial preview release of the library

README.md

@@ -21,7 +21,7 @@ import PackageDescription
 let package = Package(
     name: "signalr-client-app",
     dependencies: [
-        .package(url: "https://github.com/dotnet/signalr-client-swift", .upToNextMinor(from: "1.0.0-preview.3"))
+        .package(url: "https://github.com/dotnet/signalr-client-swift", .upToNextMinor(from: "1.0.0-preview.4"))
     ],
     targets: [
         .executableTarget(name: "YourTargetName", dependencies: [.product(name: "SignalRClient", package: "signalr-client-swift")])
@@ -113,6 +113,18 @@ for try await item in stream.stream {
 }
 ```
 
+## Client-to-server streaming
+To send a stream of data from the client to the server, use the `AsyncStream`:
+
+```swift
+let (clientStream, continuation) = AsyncStream.makeStream(of: Int.self)
+try await connection.send("UploadStream", arguments: clientStream)
+for i in 1...100 {
+    continuation.yield(i)
+}
+continuation.finish()
+``` 
+
 ## Handle lost connection
 
 ### Automatic reconnect
@@ -201,7 +213,7 @@ let connection = HubConnectionBuilder()
 | Automatic Reconnection          |✅|
 | Stateful Reconnect              ||
 | Server to Client Streaming      |✅|
-| Client to Server Streaming      ||
+| Client to Server Streaming      |✅|
 | Long Polling                    |✅|
 | Server-Sent Events              |✅|
 | WebSockets                      |✅|
@@ -211,4 +223,4 @@ let connection = HubConnectionBuilder()
 
 # License
 
-.NET (including the runtime repo) is licensed under the MIT license.
+.NET (including the runtime repo) is licensed under the MIT license.

Sources/SignalRClient/HttpConnection.swift

@@ -28,6 +28,8 @@ public struct HttpConnectionOptions {
     var webSocket: AnyObject? // Placeholder for WebSocket type
     var eventSource: EventSourceAdaptor?
     var useStatefulReconnect: Bool? // Not supported yet
+    
+    public init() {}
 }
 
 // MARK: - Models

Sources/SignalRClient/HubConnection.swift

@@ -6,6 +6,8 @@ import Foundation
 public actor HubConnection {
     private static let defaultTimeout: TimeInterval = 30
     private static let defaultPingInterval: TimeInterval = 15
+    private static let defaultStatefulReconnectBufferSize: Int = 100_000_000 // bytes of messages
+
     private var invocationBinder: DefaultInvocationBinder
     private var invocationHandler: InvocationHandler
 
@@ -17,6 +19,7 @@ public actor HubConnection {
     private let retryPolicy: RetryPolicy
     private let keepAliveScheduler: TimeScheduler
     private let serverTimeoutScheduler: TimeScheduler
+    private let statefulReconnectBufferSize: Int
 
     private var connectionStarted: Bool = false
     private var receivedHandshakeResponse: Bool = false
@@ -39,9 +42,12 @@ public actor HubConnection {
                   hubProtocol: HubProtocol,
                   retryPolicy: RetryPolicy,
                   serverTimeout: TimeInterval?,
-                  keepAliveInterval: TimeInterval?) {
+                  keepAliveInterval: TimeInterval?,
+                  statefulReconnectBufferSize: Int?) {
         self.serverTimeout = serverTimeout ?? HubConnection.defaultTimeout
         self.keepAliveInterval = keepAliveInterval ?? HubConnection.defaultPingInterval
+        self.statefulReconnectBufferSize = statefulReconnectBufferSize ?? HubConnection.defaultStatefulReconnectBufferSize
+
         self.logger = logger
         self.retryPolicy = retryPolicy
 
@@ -107,29 +113,77 @@ public actor HubConnection {
     }
 
     public func send(method: String, arguments: Any...) async throws {
-        let invocationMessage = InvocationMessage(target: method, arguments: AnyEncodableArray(arguments), streamIds: nil, headers: nil, invocationId: nil)
+        let (nonstreamArguments, streamArguments) = splitStreamArguments(arguments: arguments)
+        let streamIds = await invocationHandler.createClientStreamIds(count: streamArguments.count)
+        let invocationMessage = InvocationMessage(target: method, arguments: AnyEncodableArray(nonstreamArguments), streamIds: streamIds, headers: nil, invocationId: nil)
         let data = try hubProtocol.writeMessage(message: invocationMessage)
         logger.log(level: .debug, message: "Sending message to target: \(method)")
         try await sendMessageInternal(data)
+        launchStreams(streamIds: streamIds, clientStreams: streamArguments)
+    }
+    
+    private func splitStreamArguments(arguments: Any...) -> ([Any], [any AsyncSequence]) {
+        var nonstreamArguments: [Any] = []
+        var streamArguments: [any AsyncSequence] = []
+        for argument in arguments {
+            if let stream = argument as? (any AsyncSequence) {
+                streamArguments.append(stream)
+            } else {
+                nonstreamArguments.append(argument)
+            }
+        }
+        return (nonstreamArguments, streamArguments)
     }
 
+    private func launchStreams(streamIds: [String], clientStreams: [any AsyncSequence]) {
+        for i in 0 ..< streamIds.count {
+            Task {
+                let stream = clientStreams[i]
+                var err: String? = nil
+                do {
+                    for try await item in stream {
+                        let streamItem = StreamItemMessage(invocationId: streamIds[i], item: AnyEncodable(item), headers: nil)
+                        let data = try hubProtocol.writeMessage(message: streamItem)
+                        try await sendMessageInternal(data)
+                    }
+                } catch {
+                    err = "\(error)"
+                    logger.log(level: .error, message: "Fail to send client stream message :\(error)")
+                }
+                do {
+                    let completionMessage = CompletionMessage(invocationId: streamIds[i], error: err, result: AnyEncodable(nil), headers: nil)
+                    let data = try hubProtocol.writeMessage(message: completionMessage)
+                    try await sendMessageInternal(data)
+                } catch {
+                    logger.log(level: .error, message: "Fail to send client stream complete message :\(error)")
+                }
+            }
+        }
+    }
+    
     public func invoke(method: String, arguments: Any...) async throws -> Void {
+        let (nonstreamArguments, streamArguments) = splitStreamArguments(arguments: arguments)
+        let streamIds = await invocationHandler.createClientStreamIds(count: streamArguments.count)
         let (invocationId, tcs) = await invocationHandler.create()
-        let invocationMessage = InvocationMessage(target: method, arguments: AnyEncodableArray(arguments), streamIds: nil, headers: nil, invocationId: invocationId)
+        let invocationMessage = InvocationMessage(target: method, arguments: AnyEncodableArray(nonstreamArguments), streamIds: streamIds, headers: nil, invocationId: invocationId)
         let data = try hubProtocol.writeMessage(message: invocationMessage)
         logger.log(level: .debug, message: "Invoke message to target: \(method), invocationId: \(invocationId)")
         try await sendMessageInternal(data)
+        launchStreams(streamIds: streamIds, clientStreams: streamArguments)
         _ = try await tcs.task()
     }
 
     public func invoke<TReturn>(method: String, arguments: Any...) async throws -> TReturn {
+        let (nonstreamArguments, streamArguments) = splitStreamArguments(arguments: arguments)
+        let streamIds = await invocationHandler.createClientStreamIds(count: streamArguments.count)
         let (invocationId, tcs) = await invocationHandler.create()
         invocationBinder.registerReturnValueType(invocationId: invocationId, types: TReturn.self)
-        let invocationMessage = InvocationMessage(target: method, arguments: AnyEncodableArray(arguments), streamIds: nil, headers: nil, invocationId: invocationId)
+        let invocationMessage = InvocationMessage(target: method, arguments: AnyEncodableArray(nonstreamArguments), streamIds: streamIds, headers: nil, invocationId: invocationId)
         do {
             let data = try hubProtocol.writeMessage(message: invocationMessage)
             logger.log(level: .debug, message: "Invoke message to target: \(method), invocationId: \(invocationId)")
             try await sendMessageInternal(data)
+            launchStreams(streamIds: streamIds, clientStreams: streamArguments)
         } catch {
             await invocationHandler.cancel(invocationId: invocationId, error: error)
             invocationBinder.removeReturnValueType(invocationId: invocationId)
@@ -144,13 +198,16 @@ public actor HubConnection {
     }
 
     public func stream<Element>(method: String, arguments: Any...) async throws -> any StreamResult<Element> {
+        let (nonstreamArguments, streamArguments) = splitStreamArguments(arguments: arguments)
+        let streamIds = await invocationHandler.createClientStreamIds(count: streamArguments.count)
         let (invocationId, stream) = await invocationHandler.createStream()
         invocationBinder.registerReturnValueType(invocationId: invocationId, types: Element.self)
-        let StreamInvocationMessage = StreamInvocationMessage(invocationId: invocationId, target: method, arguments: AnyEncodableArray(arguments), streamIds: nil, headers: nil)
+        let StreamInvocationMessage = StreamInvocationMessage(invocationId: invocationId, target: method, arguments: AnyEncodableArray(nonstreamArguments), streamIds: streamIds, headers: nil)
         do {
             let data = try hubProtocol.writeMessage(message: StreamInvocationMessage)
             logger.log(level: .debug, message: "Stream message to target: \(method), invocationId: \(invocationId)")
             try await sendMessageInternal(data)
+            launchStreams(streamIds: streamIds, clientStreams: streamArguments)
         } catch {
             await invocationHandler.cancel(invocationId: invocationId, error: error)
             invocationBinder.removeReturnValueType(invocationId: invocationId)
@@ -672,6 +729,14 @@ public actor HubConnection {
             }
             return (id, stream)
         }
+        
+        func createClientStreamIds(count: Int) -> [String] {
+            var streamIds: [String] = []
+            for _ in 0 ..< count {
+                streamIds.append(nextId())
+            }
+            return streamIds
+        }
 
         func setResult(message: CompletionMessage) async {
             if let invocation = invocations[message.invocationId!] {

Sources/SignalRClient/HubConnectionBuilder.swift

@@ -12,6 +12,7 @@ public class HubConnectionBuilder {
     private var keepAliveInterval: TimeInterval?
     private var url: String?
     private var retryPolicy: RetryPolicy?
+    private var statefulReconnectBufferSize: Int?
     private var httpConnectionOptions: HttpConnectionOptions = HttpConnectionOptions()
 
     public init() {}
@@ -79,6 +80,16 @@ public class HubConnectionBuilder {
         return self
     }
 
+//    public func withStatefulReconnect() -> HubConnectionBuilder {
+//        return withStatefulReconnect(options: StatefulReconnectOptions())
+//    }
+//
+//    public func withStatefulReconnect(options: StatefulReconnectOptions) -> HubConnectionBuilder {
+//        self.statefulReconnectBufferSize = options.bufferSize
+//        self.httpConnectionOptions.useStatefulReconnect = true
+//        return self
+//    }
+
     public func build() -> HubConnection {
         guard let url = url else {
             fatalError("url must be set with .withUrl(String:)")
@@ -94,7 +105,8 @@ public class HubConnectionBuilder {
                              hubProtocol: hubProtocol,
                              retryPolicy: retryPolicy,
                              serverTimeout: serverTimeout,
-                             keepAliveInterval: keepAliveInterval)
+                             keepAliveInterval: keepAliveInterval,
+                             statefulReconnectBufferSize: statefulReconnectBufferSize)
     }
 }