Is this Stream Storage Engine implementation correct?

[darcyrush]

I want access to the ReadStream stream from Multer to pass through my application to then consume somewhere later in my application.

From the README.md link, I followed a very basic implementation to simply expose the ReadStream from file.stream. I removed all error handling and stream cleanup to provide a terse example.

I am having issues (might be NestJS framework related, or it might be my implementation below). Can anyone confirm the success callback works if simply providing the internal file.stream object is enough, as below?

class StreamEngine implements StorageEngine {
  _handleFile(
    req: Request,
    file: Express.Multer.File,
    callback: (error: Error | null, info?: Partial<AsyncFile>) => void,
  ): void {
    // Not sure a passThrough is even needed
    const passThrough = new PassThrough();
    file.stream.pipe(passThrough);

    // Execution reaches this callback, but the execution 
    // doesn't continue later in the NestJS framework controller
    callback(null, { fieldname: file.fieldname, stream: passThrough });
  })
  _removeFile(
    req: Request,
    file: AsyncFile,
    callback: (error: Error | null) => void,
  ): void {
    callback(null);
  }
}

In my case, once Multer reaches the success callback the application doesn't continue (In my NestJS framework controller).

(I'm not even sure a passThrough has any benefit given I am not transforming the stream at all but without it I have the same issue.)

In my application I pass the Readable to another library (form-data) to consume the stream. My implementation after receiving the express file object seems correct since I tested by manually creating a test ReadStream via Readable.from(createReadStream("./50Mbfile") and passing that through to the form-data library and it works as expected.

I am still working on a minimal reproduction example without an additional framework like NestJS.

[apsantiso]

I have the same problem. It seems Multer needs the stream to be consume to call the callback or something like that. Anyway, I thought it could work by doing almost the same custom engine you did but had no luck :/. It seems we'll need use busboy directly

[apsantiso]

For anyone using NestJS and having the same problem, here it is something that might work using Multer directly. Adding a body DTO won't work as usual, NestJS does not parse the body for multipart requests without buffer. You can still pass data through headers and query params if you want tho. Not sure if there's a better solution out there.

Note: Not the most clear one, as I would LOVE to be able to use FileUpload interceptor for this to follow NestJS standards, but if I use the decorator the request never reaches the controller.

  @Post('test')
  @Public()
  async uploadFile(@Req() req: Request) {
    console.log('Request received');
    const passThrough = new PassThrough();

    // Let's create a multer instance with custom storage engine
    const multerUpload = multer({
      storage: {
        _handleFile: (req, file, cb) => {
          console.log('Handling file in custom storage engine');

          file.stream.on('error', (err) => {
            console.error('Original stream error:', err);
            passThrough.emit('error', err);
          });

          file.stream.pipe(passThrough);

          cb(null, {
            stream: passThrough,
            fieldname: file.fieldname,
            originalname: file.originalname,
            encoding: file.encoding,
            mimetype: file.mimetype,
          });
        },
        _removeFile: (req, file, cb) => {
          // Nothing to remove since we're not storing anything
          cb(null);
        },
      },
    }).single('file');

    // You can use AWAIT new Promise too, but you need to make sure to pipe the stream before awaiting!
    const promiseTest = new Promise((resolve, reject) => {
      multerUpload(req, req.res, async (err) => {
        console.log('Upload finished');
        if (err) {
          console.error('Multer error:', err);
          reject(err);
        }
        resolve('Upload finished');
      });
    });

    // Make sure to add this event listener before awaiting for the promise
    passThrough.on('data', (chunk) => {
      // Pipe with another stream or process the chunk?
    });

    await promiseTest;

    return { message: 'Upload finished' };
  }

[darcyrush]

I just tested this and it works - so I assume this is indeed an issue with the NestJS framework specifically

[darcyrush]

Actually upon further testing I have the same issue as my original code - For files under 1mb, the code works, but anything over 1mb in size causes issues...

[vabatta]

In NestJS, it is not possible to directly stream multipart file uploads into a controller method when using multer or similar interceptors. The reason is that before a request reaches the controller, the interceptor must parse the incoming multipart request. However, parsing multipart uploads with multer requires an active upstream consumer to continuously read from the request stream. If the intended consumer of the stream resides within the controller itself, a deadlock occurs: the interceptor cannot finish parsing without a consumer, but the consumer cannot be invoked until the interceptor completes and the controller is executed.

This behavior can appear to “work” with small files due to Node.js stream buffering mechanics. Specifically, if the file size is smaller than the internal stream’s highWaterMark threshold, the entire file can be buffered in memory without triggering backpressure. As a result, the file seems to flow through to the interceptor and controller without issue. For larger files, however, backpressure is applied, requiring an active downstream consumer—at which point the deadlock manifests.

Example:
Suppose you configure an interceptor in NestJS with multer for handling multipart form data:

@Post('upload')
@UseInterceptors(FileInterceptor('file'))
uploadFile(@UploadedFile() file: Express.Multer.File) {
  // Intended to process file as a stream
}

• For a small file (e.g., 100 KB, below the highWaterMark), the entire content can be buffered and passed into multer, allowing the controller method to be invoked.
• For a large file (e.g., 50 MB), the Node.js stream applies backpressure because the buffered chunks exceed the highWaterMark. At this point, multer waits for a consumer, but the controller has not yet been called—creating a deadlock in the request pipeline.

I’m currently working on a solution package to address this limitation by enabling true streaming multipart handling within NestJS controllers.

[darcyrush]

@vabatta Thank you kindly for the detailed explanation!

Good luck with your implementation. I still have this need, so if you need testers, please just ping my github handle and I'll be glad to test if you get something working.

[vabatta]

@darcyrush You can follow the experiment I'm doing here: proventuslabs/nestjs#17

[vabatta]

@darcyrush I just release an alpha version if you want to check it out: https://www.npmjs.com/package/@proventuslabs/nestjs-multipart-form

[niksy]

Having same issue by just trying to consume file.stream as readable stream in FormData object. Saving to disk with disk storage engine and then reading file as stream works as expected.

Using PassThrough works for small files, but for larger files it hangs and won’t finish.

Not using any framework, just Express with middleware.

What is the correct approach here if we just want to consume stream in middleware after Multer?

[vabatta]

I'm not entirely sure on how multer works in a "pure" express setup (nest is just built on top of it), but you should be able to achieve it by writing your custom storage engine to attach the incoming file stream into the request and read it downstream in your next middleware / handler.

[niksy]

This is what I’m doing based on original post custom engine implementation, but it doesn’t work.

[vabatta]

Then it won't be possible; you will have to write your own multer middleware by hand as the only way for it to work is to let a downstream consumer attach to the file stream(s) (see #1291 (comment) for the explanation).