How do I stream a chunked file using Node.js Readable?

user_2272048 0
Viewed 560

I have a 400Mb file split into chunks that are ~1Mb each.

Each chunk is a MongoDB document:

{
  name: 'stuff.zip',
  index: 15,
  buffer: Binary('......'),
  totalChunks: 400
}

I am fetching each chunk from my database and then streaming it to the client.

Every time I get chunk from the DB I push it to the readableStream which is being piped to the client.

Here is the code:

import { Readable } from 'stream'

const name = 'stuff.zip'
const contentType = 'application/zip'

app.get('/api/download-stuff', (req, res) => {
  res.set('Content-Type', contentType)
  res.set('Content-Disposition', `attachment; filename=${name}`)
  res.attachment(name)

  // get `totalChunks` from random chunk
  let { totalChunks } = await ChunkModel.findOne({ name }).select('totalChunks')

  let index = 0

  const readableStream = new Readable({
    async read() {
      if (index < totalChunks) {

        let { buffer } = await ChunkModel.findOne({ name, index }).select('buffer')
        let canContinue = readableStream.push(buffer)
        console.log(`pushed chunk ${index}/${totalChunks}`)
        index++

        // sometimes it logs false
        // which means I should be waiting before pushing more
        // but I don't know how
        console.log('canContinue = ', canContinue)

      } else {

        readableStream.push(null)
        readableStream.destroy()
        console.log(`all ${totalChunks} chunks streamed to the client`)

      }
    }
  })

  readableStream.pipe(res)
})

The code works.

But I'm wondering whether I risk having memory overflows on my local server memory, especially when the requests for the same file are too many or the chunks are too many.

Question: My code is not waiting for readableStream to finish reading the chunk that was just pushed to it, before pushing the next one. I thought it was, and that is why I'm using read(){..} in this probably wrong way. So how should I wait for each chunk to be pushed, read, streamed to the client and cleared from my server's local memory, before I push the next one in ?

I have created this sandbox in case it helps anyone

mongoose fs node.js streaming stream
1 Answers

In general, when the readable interface is implemented correctly (i.e., the backpressure signal is respected), the readable interface will prevent the code from overflowing the memory regardless of source size.

When implemented according to the API spec, the readable itself does not keep references for data that has finished passing through the stream. The memory requirement of a readable buffer is adjusted by specifying a highWatermark.

In this case, the snippet does not conform to the readable interface. It violates the following two concepts:

  1. No data shall be pushed to the readable's buffer unless read() has been called. Currently, this implementation proceeds to push data from DB immediately. Consequently, the readable buffer will start to fill before the sink has begun to consume data.
  2. The readable's push() method returns a boolean flag. When the flag is false, the implementation must wait for .read() to be called before pushing additional data. If the flag is ignored, the buffer will overflow wrt. the highWatermark.

Note that ignoring these core criteria of Readables circumvents the backpressure logic.

An alternative implementation, if this is a Mongoose query:

app.get('/api/download-stuff', async (req, res) => {
  // ... truncated handler  
  
  // A helper variable to relay data from the stream to the response body
  const passThrough = new stream.PassThrough({objectMode: false});

  // Pipe data using pipeline() to simplify handling stream errors
  stream.pipeline(
    // Create a cursor that fetch all relevant documents using a single query
    ChunkModel.find().limit(chunksLength).select("buffer").sort({index: 1}).lean().cursor(),
    // Cherry pick the `buffer` property
    new stream.Transform({
      objectMode: true,
      transform: ({ buffer }, encoding, next) => {
        next(null, buffer);
      }
    }),
    // Write the retrieved documents to the helper variable
    passThrough,
    error => {
      if(error){
        // Log and handle error. At this point the HTTP headers are probably already sent,
        // and it is therefore too late to return HTTP500
      }
    }
  );

  res.body = passThrough;
});
user_11496320
Related