testingbrecon/tsStreaming

6 files changed, 41 insertions, 8 deletions

diff --git a/src/SignalR/clients/ts/FunctionalTests/ts/Common.ts b/src/SignalR/clients/ts/FunctionalTests/ts/Common.ts
index c66bff8e49..1185dc81a0 100644
--- a/src/SignalR/clients/ts/FunctionalTests/ts/Common.ts
+++ b/src/SignalR/clients/ts/FunctionalTests/ts/Common.ts
@@ -74,6 +74,7 @@ export function getHttpTransportTypes(): HttpTransportType[] {
             transportTypes.push(HttpTransportType.ServerSentEvents);
         }
     }
+    transportTypes.push(HttpTransportType.HttpStreaming);
     transportTypes.push(HttpTransportType.LongPolling);
 
     return transportTypes;
@@ -106,6 +107,9 @@ export function eachTransportAndProtocol(action: (transport: HttpTransportType,
 export function eachTransportAndProtocolAndHttpClient(action: (transport: HttpTransportType, protocol: IHubProtocol, httpClient: HttpClient) => void) {
     eachTransportAndProtocol((transport, protocol) => {
         getHttpClients().forEach((httpClient) => {
+            if (transport === HttpTransportType.HttpStreaming && !httpClient.supportsStreaming) {
+                return;
+            }
             action(transport, protocol, httpClient);
         });
     });
diff --git a/src/SignalR/clients/ts/FunctionalTests/ts/ConnectionTests.ts b/src/SignalR/clients/ts/FunctionalTests/ts/ConnectionTests.ts
index 3b559af265..ba5ae623de 100644
--- a/src/SignalR/clients/ts/FunctionalTests/ts/ConnectionTests.ts
+++ b/src/SignalR/clients/ts/FunctionalTests/ts/ConnectionTests.ts
@@ -10,6 +10,7 @@ import { TestLogger } from "./TestLogger";
 
 // We want to continue testing HttpConnection, but we don't export it anymore. So just pull it in directly from the source file.
 import { HttpConnection } from "@microsoft/signalr/dist/esm/HttpConnection";
+import { formatArrayBuffer } from "@microsoft/signalr/dist/esm/Utils";
 import "./LogBannerReporter";
 
 const commonOptions: IHttpConnectionOptions = {
@@ -45,6 +46,9 @@ describe("connection", () => {
 
     eachTransport((transportType) => {
         eachHttpClient((httpClient) => {
+            if (transportType === HttpTransportType.HttpStreaming && !httpClient.supportsStreaming) {
+                return;
+            }
             describe(`over ${HttpTransportType[transportType]} with ${(httpClient.constructor as any).name}`, () => {
                 it("can send and receive messages", (done) => {
                     const message = "Hello World!";
@@ -57,7 +61,8 @@ describe("connection", () => {
                     });
 
                     connection.onreceive = (data: any) => {
-                        if (data === message) {
+                        if ((typeof data === "string" && data === message) ||
+                            (data instanceof ArrayBuffer && new TextDecoder("utf-8").decode(data) === message)) {
                             connection.stop();
                         }
                     };
@@ -87,7 +92,8 @@ describe("connection", () => {
                     });
 
                     connection.onreceive = (data: any) => {
-                        if (data === message) {
+                        if ((typeof data === "string" && data === message) ||
+                            (data instanceof ArrayBuffer && new TextDecoder("utf-8").decode(data) === message)) {
                             connection.stop();
                         }
                     };
@@ -114,6 +120,7 @@ describe("connection", () => {
                 it("does log content of messages sent or received when enabled", (done) => {
                     TestLogger.saveLogsAndReset();
                     const message = "Hello World!";
+                    const encodedMessage = formatArrayBuffer(new TextEncoder().encode(message).buffer);
 
                     // DON'T use commonOptions because we want to specifically test the scenario where logMessageContent is set to true (even if commonOptions changes).
                     const connection = new HttpConnection(ECHOENDPOINT_URL, {
@@ -124,7 +131,8 @@ describe("connection", () => {
                     });
 
                     connection.onreceive = (data: any) => {
-                        if (data === message) {
+                        if ((typeof data === "string" && data === message) ||
+                            (data instanceof ArrayBuffer && new TextDecoder("utf-8").decode(data) === message)) {
                             connection.stop();
                         }
                     };
@@ -138,6 +146,8 @@ describe("connection", () => {
                         for (const [_, __, logMessage] of TestLogger.instance.currentLog.messages) {
                             if (logMessage.indexOf(message) !== -1) {
                                 matches += 1;
+                            } else if (logMessage.indexOf(encodedMessage.toString()) !== -1) {
+                                matches += 1;
                             }
                         }
 
diff --git a/src/SignalR/clients/ts/signalr/src/FetchHttpClient.ts b/src/SignalR/clients/ts/signalr/src/FetchHttpClient.ts
index f77b765db5..a0e2859801 100644
--- a/src/SignalR/clients/ts/signalr/src/FetchHttpClient.ts
+++ b/src/SignalR/clients/ts/signalr/src/FetchHttpClient.ts
@@ -80,7 +80,7 @@ export class FetchHttpClient extends HttpClient {
             if (timeoutId) {
                 clearTimeout(timeoutId);
             }
-            if (request.abortSignal) {
+            if (request.abortSignal && !request.stream) {
                 request.abortSignal.onabort = null;
             }
         }
diff --git a/src/SignalR/clients/ts/signalr/src/HttpStreamingTransport.ts b/src/SignalR/clients/ts/signalr/src/HttpStreamingTransport.ts
index 2ee78281a6..cb342875df 100644
--- a/src/SignalR/clients/ts/signalr/src/HttpStreamingTransport.ts
+++ b/src/SignalR/clients/ts/signalr/src/HttpStreamingTransport.ts
@@ -1,10 +1,11 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
+import { AbortController } from "./AbortController";
 import { HttpClient, HttpResponse } from "./HttpClient";
 import { ILogger, LogLevel } from "./ILogger";
 import { ITransport, TransferFormat } from "./ITransport";
-import { getDataDetail } from "./Utils";
+import { getDataDetail, getUserAgentHeader } from "./Utils";
 
 // Not exported from 'index', this type is internal.
 /** @private */
@@ -17,6 +18,7 @@ export class HttpStreamingTransport implements ITransport {
     private streamPromise: Promise<any>;
     private url: string = "";
     private reader: ReadableStreamReader | null;
+    private streamAbort: AbortController;
 
     public onreceive: ((data: string | ArrayBuffer) => void) | null;
     public onclose: ((error?: Error) => void) | null;
@@ -34,6 +36,7 @@ export class HttpStreamingTransport implements ITransport {
         this.onclose = null;
         this.reader = null;
         this.streamPromise = Promise.resolve();
+        this.streamAbort = new AbortController();
     }
 
     // @ts-ignore
@@ -44,7 +47,10 @@ export class HttpStreamingTransport implements ITransport {
         const headers: { [key: string]: string } = {};
         this.updateHeaderToken(headers, token);
 
-        const response = await this.httpClient.get(url, { stream: true, headers });
+        const [name, value] = getUserAgentHeader();
+        headers[name] = value;
+
+        const response = await this.httpClient.get(url, { abortSignal: this.streamAbort.signal, headers, stream: true });
 
         this.streamPromise = this.stream(response);
     }
@@ -73,13 +79,23 @@ export class HttpStreamingTransport implements ITransport {
     }
 
     public async send(data: any): Promise<void> {
-        await this.httpClient.post(this.url, { content: data });
+        const token = await this.getAccessToken();
+        const headers: { [key: string]: string } = {};
+        this.updateHeaderToken(headers, token);
+
+        const [name, value] = getUserAgentHeader();
+        headers[name] = value;
+
+        this.logger.log(LogLevel.Trace, `(HttpStreaming transport) sending data. ${getDataDetail(data, this.logMessageContent)}.`);
+
+        await this.httpClient.post(this.url, { content: data, headers });
     }
 
     public async stop(): Promise<void> {
         if (this.reader) {
             await this.reader.cancel();
         }
+        this.streamAbort.abort();
         await this.streamPromise;
 
         if (this.onclose) {
diff --git a/src/SignalR/clients/ts/signalr/src/LongPollingTransport.ts b/src/SignalR/clients/ts/signalr/src/LongPollingTransport.ts
index bc92a7bfcd..7fc46ef887 100644
--- a/src/SignalR/clients/ts/signalr/src/LongPollingTransport.ts
+++ b/src/SignalR/clients/ts/signalr/src/LongPollingTransport.ts
@@ -143,7 +143,7 @@ export class LongPollingTransport implements ITransport {
                         // Process the response
                         if (response.content) {
                             this.logger.log(LogLevel.Trace, `(LongPolling transport) data received. ${getDataDetail(response.content, this.logMessageContent)}.`);
-                            if (this.onreceive && !(response.content instanceof ReadableStream)) {
+                            if (this.onreceive && (response.content instanceof ArrayBuffer || typeof response.content === "string")) {
                                 this.onreceive(response.content);
                             }
                         } else {
diff --git a/src/SignalR/clients/ts/signalr/src/Utils.ts b/src/SignalR/clients/ts/signalr/src/Utils.ts
index df833c96d9..9939481889 100644
--- a/src/SignalR/clients/ts/signalr/src/Utils.ts
+++ b/src/SignalR/clients/ts/signalr/src/Utils.ts
@@ -45,6 +45,9 @@ export class Platform {
 /** @private */
 export function getDataDetail(data: any, includeContent: boolean): string {
     let detail = "";
+    if (data instanceof Uint8Array) {
+        data = data.buffer;
+    }
     if (isArrayBuffer(data)) {
         detail = `Binary data of length ${data.byteLength}`;
         if (includeContent) {