Merge pull request #8740 from wSedlacek/feat/handle-observable-stream…

…-errors

feat(core): handle errors in observable streams

packages/core/router/router-execution-context.ts

@@ -431,7 +431,7 @@ export class RouterExecutionContext {
     }
     const isSseHandler = !!this.reflectSse(callback);
     if (isSseHandler) {
-      return async <
+      return <
         TResult extends Observable<unknown> = any,
         TResponse extends HeaderStream = any,
         TRequest extends IncomingMessage = any,
@@ -440,7 +440,7 @@ export class RouterExecutionContext {
         res: TResponse,
         req: TRequest,
       ) => {
-        await this.responseController.sse(
+        this.responseController.sse(
           result,
           (res as any).raw || res,
           (req as any).raw || req,

packages/core/router/router-response-controller.ts

@@ -1,8 +1,14 @@
-import { HttpServer, HttpStatus, RequestMethod } from '@nestjs/common';
+import {
+  HttpServer,
+  HttpStatus,
+  Logger,
+  RequestMethod,
+  MessageEvent,
+} from '@nestjs/common';
 import { isFunction, isObject } from '@nestjs/common/utils/shared.utils';
 import { IncomingMessage } from 'http';
-import { lastValueFrom, Observable } from 'rxjs';
-import { debounce } from 'rxjs/operators';
+import { EMPTY, lastValueFrom, Observable } from 'rxjs';
+import { catchError, debounce, map } from 'rxjs/operators';
 import {
   AdditionalHeaders,
   WritableHeaderStream,
@@ -20,6 +26,8 @@ export interface RedirectResponse {
 }
 
 export class RouterResponseController {
+  private readonly logger = new Logger(RouterResponseController.name);
+
   constructor(private readonly applicationRef: HttpServer) {}
 
   public async apply<TInput = any, TResponse = any>(
@@ -87,7 +95,7 @@ export class RouterResponseController {
     this.applicationRef.status(response, statusCode);
   }
 
-  public async sse<
+  public sse<
     TInput extends Observable<unknown> = any,
     TResponse extends WritableHeaderStream = any,
     TRequest extends IncomingMessage = any,
@@ -109,15 +117,29 @@ export class RouterResponseController {
 
     const subscription = result
       .pipe(
+        map((message): MessageEvent => {
+          if (isObject(message)) {
+            return message as MessageEvent;
+          }
+
+          return { data: message as object | string };
+        }),
         debounce(
-          (message: any) =>
-            new Promise(resolve => {
-              if (!isObject(message)) {
-                message = { data: message };
-              }
-              stream.writeMessage(message, resolve);
-            }),
+          message =>
+            new Promise<void>(resolve =>
+              stream.writeMessage(message, () => resolve()),
+            ),
         ),
+        catchError(err => {
+          const data = err instanceof Error ? err.message : err;
+          stream.writeMessage({ type: 'error', data }, writeError => {
+            if (writeError) {
+              this.logger.error(writeError);
+            }
+          });
+
+          return EMPTY;
+        }),
       )
       .subscribe({
         complete: () => {

packages/core/test/router/router-response-controller.spec.ts

@@ -1,7 +1,7 @@
 import { isNil, isObject } from '@nestjs/common/utils/shared.utils';
 import { expect } from 'chai';
 import { IncomingMessage, ServerResponse } from 'http';
-import { Observable, of } from 'rxjs';
+import { Observable, of, Subject } from 'rxjs';
 import * as sinon from 'sinon';
 import { PassThrough, Writable } from 'stream';
 import { HttpStatus, RequestMethod } from '../../../common';
@@ -349,14 +349,86 @@ data: test
       const request = new Writable();
       request._write = () => {};
 
-      routerResponseController.sse(
-        result as unknown as Observable<string>,
-        response as unknown as ServerResponse,
-        request as unknown as IncomingMessage,
-      );
+      try {
+        routerResponseController.sse(
+          result as unknown as Observable<string>,
+          response as unknown as ServerResponse,
+          request as unknown as IncomingMessage,
+        );
+      } catch {
+        // Wether an error is thrown or not
+        // is not relevant, so long as
+        // result is not called
+      }
 
       sinon.assert.notCalled(result);
       done();
     });
+
+    describe('when there is an error', () => {
+      it('should close the request', done => {
+        const result = new Subject();
+        const response = new Writable();
+        response.end = done;
+        response._write = () => {};
+
+        const request = new Writable();
+        request._write = () => {};
+
+        routerResponseController.sse(
+          result,
+          response as unknown as ServerResponse,
+          request as unknown as IncomingMessage,
+        );
+
+        result.error(new Error('Some error'));
+      });
+
+      it('should write the error message to the stream', async () => {
+        class Sink extends Writable {
+          private readonly chunks: string[] = [];
+
+          _write(
+            chunk: any,
+            encoding: string,
+            callback: (error?: Error | null) => void,
+          ): void {
+            this.chunks.push(chunk);
+            callback();
+          }
+
+          get content() {
+            return this.chunks.join('');
+          }
+        }
+
+        const written = (stream: Writable) =>
+          new Promise((resolve, reject) =>
+            stream.on('error', reject).on('finish', resolve),
+          );
+
+        const result = new Subject();
+        const response = new Sink();
+        const request = new PassThrough();
+        routerResponseController.sse(
+          result,
+          response as unknown as ServerResponse,
+          request as unknown as IncomingMessage,
+        );
+
+        result.error(new Error('Some error'));
+        request.destroy();
+
+        await written(response);
+        expect(response.content).to.eql(
+          `:
+event: error
+id: 1
+data: Some error
+
+`,
+        );
+      });
+    });
   });
 });