Merge pull request #9819 from jmcdo29/fix/pipeline-over-pipe

fix: use pipeline over stream.pipe

Author: kamilmysliwiec

integration/send-files/e2e/express.spec.ts

@@ -65,4 +65,7 @@ describe('Express FileSend', () => {
         expect(res.text).to.be.eq(readmeString);
       });
   });
+  it('should return an error if the file does not exist', async () => {
+    return request(app.getHttpServer()).get('/file/not/exist').expect(400);
+  });
 });

integration/send-files/src/app.controller.ts

@@ -31,4 +31,9 @@ export class AppController {
   getFileWithHeaders(): StreamableFile {
     return this.appService.getFileWithHeaders();
   }
+
+  @Get('file/not/exist')
+  getNonExistantFile(): StreamableFile {
+    return this.appService.getFileThatDoesNotExist();
+  }
 }

integration/send-files/src/app.service.ts

@@ -35,4 +35,8 @@ export class AppService {
       },
     );
   }
+
+  getFileThatDoesNotExist(): StreamableFile {
+    return new StreamableFile(createReadStream('does-not-exist.txt'));
+  }
 }

packages/common/file-stream/streamable-file.ts

@@ -3,9 +3,22 @@ import { types } from 'util';
 import { isFunction } from '../utils/shared.utils';
 import { StreamableFileOptions } from './streamable-options.interface';
 
+export interface StreamableHandlerResponse {
+  statusCode: number;
+  send: (msg: string) => void;
+}
+
 export class StreamableFile {
   private readonly stream: Readable;
 
+  protected handleError: (
+    err: Error,
+    response: StreamableHandlerResponse,
+  ) => void = (err: Error, res) => {
+    res.statusCode = 400;
+    res.send(err.message);
+  };
+
   constructor(buffer: Uint8Array, options?: StreamableFileOptions);
   constructor(readable: Readable, options?: StreamableFileOptions);
   constructor(
@@ -38,4 +51,18 @@ export class StreamableFile {
       length,
     };
   }
+
+  get errorHandler(): (
+    err: Error,
+    response: StreamableHandlerResponse,
+  ) => void {
+    return this.handleError;
+  }
+
+  setErrorHandler(
+    handler: (err: Error, response: StreamableHandlerResponse) => void,
+  ) {
+    this.handleError = handler;
+    return this;
+  }
 }

packages/platform-express/adapters/express-adapter.ts

@@ -1,5 +1,6 @@
 import {
   InternalServerErrorException,
+  Logger,
   RawBodyRequest,
   RequestMethod,
   StreamableFile,
@@ -32,6 +33,7 @@ import * as cors from 'cors';
 import * as express from 'express';
 import * as http from 'http';
 import * as https from 'https';
+import { PassThrough, pipeline } from 'stream';
 import { ServeStaticOptions } from '../interfaces/serve-static-options.interface';
 import { getBodyParserOptions } from './utils/get-body-parser-options.util';
 
@@ -46,6 +48,7 @@ type VersionedRoute = <
 
 export class ExpressAdapter extends AbstractHttpAdapter {
   private readonly routerMethodFactory = new RouterMethodFactory();
+  private readonly logger = new Logger(ExpressAdapter.name);
 
   constructor(instance?: any) {
     super(instance || express());
@@ -78,7 +81,17 @@ export class ExpressAdapter extends AbstractHttpAdapter {
       ) {
         response.setHeader('Content-Length', streamHeaders.length);
       }
-      return body.getStream().pipe(response);
+      return pipeline(
+        body.getStream().once('error', (err: Error) => {
+          body.errorHandler(err, response);
+        }),
+        response,
+        (err: Error) => {
+          if (err) {
+            this.logger.error(err.message, err.stack);
+          }
+        },
+      );
     }
     return isObject(body) ? response.json(body) : response.send(String(body));
   }