fix(core): possible memory leak when using server side events

Author: zhengjitf

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

@@ -8,7 +8,7 @@ import {
 import { isObject } from '@nestjs/common/utils/shared.utils';
 import { IncomingMessage } from 'http';
 import { EMPTY, lastValueFrom, Observable, isObservable } from 'rxjs';
-import { catchError, debounce, map } from 'rxjs/operators';
+import { catchError, concatMap, map } from 'rxjs/operators';
 import {
   AdditionalHeaders,
   WritableHeaderStream,
@@ -128,7 +128,7 @@ export class RouterResponseController {
 
           return { data: message as object | string };
         }),
-        debounce(
+        concatMap(
           message =>
             new Promise<void>(resolve =>
               stream.writeMessage(message, () => resolve()),
@@ -153,6 +153,9 @@ export class RouterResponseController {
 
     request.on('close', () => {
       subscription.unsubscribe();
+      if (!stream.writableEnded) {
+        stream.end();
+      }
     });
   }
 

packages/core/router/sse-stream.ts

@@ -116,7 +116,7 @@ export class SseStream extends Transform {
       message.id = this.lastEventId.toString();
     }
 
-    if (!this.write(message, 'utf-8', cb)) {
+    if (!this.write(message, 'utf-8')) {
       this.once('drain', cb);
     } else {
       process.nextTick(cb);

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

@@ -436,9 +436,7 @@ data: test
 
         await new Promise(resolve => process.nextTick(resolve));
 
-        expect(() => {
-          expect(maxDrainListenersExceededWarning).to.equal(null);
-        }, 'it will fail as there is an issue here to be addressed').to.throw();
+        expect(maxDrainListenersExceededWarning).to.equal(null);
       });
     });