[SignalR TS] Ignore user errors from stream callbacks (#34136)

Author: BrennanConroy

src/SignalR/clients/ts/signalr/src/HubConnection.ts

@@ -8,7 +8,7 @@ import { ILogger, LogLevel } from "./ILogger";
 import { IRetryPolicy } from "./IRetryPolicy";
 import { IStreamResult } from "./Stream";
 import { Subject } from "./Subject";
-import { Arg } from "./Utils";
+import { Arg, getErrorString } from "./Utils";
 
 const DEFAULT_TIMEOUT_IN_MS: number = 30 * 1000;
 const DEFAULT_PING_INTERVAL_IN_MS: number = 15 * 1000;
@@ -544,7 +544,11 @@ export class HubConnection {
                             if (message.type === MessageType.Completion) {
                                 delete this._callbacks[message.invocationId];
                             }
-                            callback(message);
+                            try {
+                                callback(message);
+                            } catch (e) {
+                                this._logger.log(LogLevel.Error, `Stream callback threw error: ${getErrorString(e)}`);
+                            }
                         }
                         break;
                     }
@@ -830,7 +834,11 @@ export class HubConnection {
         Object.keys(callbacks)
             .forEach((key) => {
                 const callback = callbacks[key];
-                callback(null, error);
+                try {
+                    callback(null, error);
+                } catch (e) {
+                    this._logger.log(LogLevel.Error, `Stream 'error' callback called with '${error}' threw error: ${getErrorString(e)}`);
+                }
             });
     }
 

src/SignalR/clients/ts/signalr/src/Utils.ts

@@ -266,3 +266,13 @@ function getRuntime(): string {
         return "Browser";
     }
 }
+
+/** @private */
+export function getErrorString(e: any): string {
+    if (e.stack) {
+        return e.stack;
+    } else if (e.message) {
+        return e.message;
+    }
+    return `${e}`;
+}

src/SignalR/clients/ts/signalr/tests/HubConnection.test.ts

@@ -1190,6 +1190,57 @@ describe("HubConnection", () => {
                 }
             });
         });
+
+        it("ignores error from 'next' and 'complete' function", async () => {
+            await VerifyLogger.run(async (logger) => {
+                const connection = new TestConnection();
+                const hubConnection = createHubConnection(connection, logger);
+                try {
+                    await hubConnection.start();
+
+                    hubConnection.stream("testMethod")
+                        .subscribe({
+                            next: (_item) => {
+                                throw new Error("from next");
+                            },
+                            error: (_e) => {},
+                            complete: () => {
+                                throw new Error("from complete");
+                            }
+                        });
+
+                    connection.receive({ type: MessageType.StreamItem, invocationId: connection.lastInvocationId, item: 1 });
+
+                    connection.receive({ type: MessageType.Completion, invocationId: connection.lastInvocationId });
+                } finally {
+                    await hubConnection.stop();
+                }
+            }, /Stream callback threw error: Error: from complete/,
+            /Stream callback threw error: Error: from next/);
+        });
+
+        it("ignores error from 'error' function", async () => {
+            await VerifyLogger.run(async (logger) => {
+                const connection = new TestConnection();
+                const hubConnection = createHubConnection(connection, logger);
+                try {
+                    await hubConnection.start();
+
+                    hubConnection.stream("testMethod")
+                        .subscribe({
+                            next: (_item) => {},
+                            error: (_e) => {
+                                throw new Error("from error");
+                            },
+                            complete: () => {}
+                        });
+
+                    connection.receive({ type: MessageType.StreamItem, invocationId: connection.lastInvocationId, item: 1 });
+                } finally {
+                    await hubConnection.stop();
+                }
+            }, /Stream 'error' callback called with 'Error: Invocation canceled due to the underlying connection being closed.' threw error: Error: from error/);
+        });
     });
 
     describe("onClose", () => {