Changing streaming things

Author: BrennanConroy

clients/ts/signalr/src/UploadStream.ts

@@ -3,6 +3,7 @@
 
 using System;
 using System.Reactive.Linq;
+using System.Text;
 using System.Threading.Channels;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http.Connections;
@@ -50,6 +51,21 @@ public ChannelReader<string> Stream()
             return channel.Reader;
         }
 
+        public async Task<string> StreamingConcat(ChannelReader<string> stream)
+        {
+            var sb = new StringBuilder();
+
+            while (await stream.WaitToReadAsync())
+            {
+                while (stream.TryRead(out var item))
+                {
+                    sb.Append(item);
+                }
+            }
+
+            return sb.ToString();
+        }
+
         public ChannelReader<int> EmptyStream()
         {
             var channel = Channel.CreateUnbounded<int>();

src/SignalR/clients/ts/FunctionalTests/TestHub.cs

@@ -30,6 +30,7 @@
     "karma-sauce-launcher": "^1.2.0",
     "karma-sourcemap-loader": "^0.3.7",
     "karma-summary-reporter": "^1.5.0",
+    "rxjs": "^6.3.3",
     "ts-node": "^4.1.0",
     "typescript": "^3.0.1",
     "ws": " ^6.0.0"

src/SignalR/clients/ts/FunctionalTests/package-lock.json

@@ -11,6 +11,8 @@ import { eachTransport, eachTransportAndProtocol, ENDPOINT_BASE_HTTPS_URL, ENDPO
 import "./LogBannerReporter";
 import { TestLogger } from "./TestLogger";
 
+import * as RX from "rxjs";
+
 const TESTHUBENDPOINT_URL = ENDPOINT_BASE_URL + "/testhub";
 const TESTHUBENDPOINT_HTTPS_URL = ENDPOINT_BASE_HTTPS_URL ? (ENDPOINT_BASE_HTTPS_URL + "/testhub") : undefined;
 
@@ -531,6 +533,46 @@ describe("hubConnection", () => {
                     done();
                 });
             });
+
+            it("can stream from client to server with rxjs", async (done) => {
+                const hubConnection = getConnectionBuilder(transportType)
+                    .withHubProtocol(protocol)
+                    .build();
+
+                await hubConnection.start();
+                const subject = new RX.Subject<string>();
+                const resultPromise = hubConnection.invoke<string>("StreamingConcat", subject.asObservable());
+                subject.next("Hello ");
+                subject.next("world");
+                subject.next("!");
+                subject.complete();
+                expect(await resultPromise).toBe("Hello world!");
+                await hubConnection.stop();
+                done();
+            });
+
+            it("can stream from client to server and close with error with rxjs", async (done) => {
+                const hubConnection = getConnectionBuilder(transportType)
+                    .withHubProtocol(protocol)
+                    .build();
+
+                await hubConnection.start();
+                const subject = new RX.Subject<string>();
+                const resultPromise = hubConnection.invoke<string>("StreamingConcat", subject.asObservable());
+                subject.next("Hello ");
+                subject.next("world");
+                subject.next("!");
+                subject.error(new Error("Something bad"));
+                try {
+                    await resultPromise;
+                    expect(false).toBe(true);
+                } catch (err) {
+                    expect(err.message).toEqual("An unexpected error occurred invoking 'StreamingConcat' on the server. Exception: Something bad");
+                } finally {
+                    await hubConnection.stop();
+                }
+                done();
+            });
         });
     });
 

src/SignalR/clients/ts/FunctionalTests/package.json

@@ -4,7 +4,7 @@
 import { Buffer } from "buffer";
 import * as msgpack5 from "msgpack5";
 
-import { CompletionMessage, HubMessage, IHubProtocol, ILogger, InvocationMessage, LogLevel, MessageHeaders, MessageType, NullLogger, StreamInvocationMessage, StreamItemMessage, TransferFormat } from "@aspnet/signalr";
+import { CompletionMessage, HubMessage, IHubProtocol, ILogger, InvocationMessage, LogLevel, MessageHeaders, MessageType, NullLogger, StreamCompleteMessage, StreamDataMessage, StreamInvocationMessage, StreamItemMessage, TransferFormat } from "@aspnet/signalr";
 
 import { BinaryMessageFormat } from "./BinaryMessageFormat";
 import { isArrayBuffer } from "./Utils";
@@ -65,11 +65,15 @@ export class MessagePackHubProtocol implements IHubProtocol {
                 return this.writeInvocation(message as InvocationMessage);
             case MessageType.StreamInvocation:
                 return this.writeStreamInvocation(message as StreamInvocationMessage);
+            case MessageType.StreamData:
+                return this.writeStreamData(message as StreamDataMessage);
             case MessageType.StreamItem:
             case MessageType.Completion:
                 throw new Error(`Writing messages of type '${message.type}' is not supported.`);
             case MessageType.Ping:
                 return BinaryMessageFormat.write(SERIALIZED_PING_MESSAGE);
+            case MessageType.StreamComplete:
+                return this.writeStreamComplete(message as StreamCompleteMessage);
             default:
                 throw new Error("Invalid message type.");
         }
@@ -226,6 +230,22 @@ export class MessagePackHubProtocol implements IHubProtocol {
         return BinaryMessageFormat.write(payload.slice());
     }
 
+    private writeStreamData(streamDataMessage: StreamDataMessage): ArrayBuffer {
+        const msgpack = msgpack5();
+        const payload = msgpack.encode([MessageType.StreamData, streamDataMessage.streamId,
+            streamDataMessage.item]);
+
+        return BinaryMessageFormat.write(payload.slice());
+    }
+
+    private writeStreamComplete(streamCompleteMessage: StreamCompleteMessage): ArrayBuffer {
+        const msgpack = msgpack5();
+        const payload = msgpack.encode([MessageType.StreamComplete, streamCompleteMessage.streamId,
+            streamCompleteMessage.error || null]);
+
+        return BinaryMessageFormat.write(payload.slice());
+    }
+
     private readHeaders(properties: any): MessageHeaders {
         const headers: MessageHeaders = properties[1] as MessageHeaders;
         if (typeof headers !== "object") {