How to defer stream read invocation

Question

I'm still trying to grok my way through streams in general. I have been able to stream a large file using multiparty from within form.on('part'). But I need to defer the invocation and resolve the stream before it's read. I have tried PassThrough, through. through2, but have gotten different results, which it mainly hangs, and I can't figure out what to do, nor steps to debug. I'm open to all alternatives. Thanks for all insights.

import multiparty from 'multiparty'
import {
  PassThrough
} from 'stream';
import through from 'through'
import through2 from 'through2'

export function promisedMultiparty(req) {
  return new Promise((resolve, reject) => {

    const form = new multiparty.Form()
    const form_files = []
    let q_str = ''

    form.on('field', (fieldname, value) => {
      if (value) q_str = appendQStr(fieldname, value, q_str)
    })

    form.on('part', async (part) => {
      if (part.filename) {

        const pass1 = new PassThrough() // this hangs at 10% 

        const pass2 = through(function write(data) { // this hangs from the beginning
            this.queue(data)
          },
          function end() {
            this.queue(null)
          })

        const pass3 = through2() // this hangs at 10%

        /* 
            // This way works for large files, but I want to defer 
            // invocation

            const form_data = new FormData()
            form_data.append(savepath, part, {
              filename,
            })

            const r = request.post(url, {
              headers: {
                'transfer-encoding': 'chunked'
              }
            }, responseCallback(resolve))
            r._form = form

        */

        form_files.push({
          part: part.pipe(pass1),
          // part: part.pipe(pass2),
          // part: part.pipe(pass3),
        })

      } else {
        part.resume()
      }
    })

    form.on('close', () => {
      resolve({
        fields: qs.parse(q_str),
        forms: form_files,
      })
    })

    form.parse(req)
  })
}

p.s. For sure the title could be better, if someone could use the proper terms please. Thanks.

I should point out that `PassThrough` and `through2` works with smaller files. Does `PassThrough` and `through2` behave the same way (hangs at 10%) because they are based on `Stream2`? — garajo, Sep 13 '18 at 19:21
Thanks for having a gander @F.bernal. Instead of sending the request from within `form.on('part')`, I want to resolve with the unread stream, and in the function where the promise resolves, start streaming with additional context. — garajo, Sep 21 '18 at 09:00
Any tips on how to debug or expose the stream post `multiparty` would be great otherwise. — garajo, Sep 21 '18 at 09:09
Correct me if I'm wrong. You want that multipart do its job by giving you the file's chunks and the point where your function promisedMultiparty is called start doing the file streaming? — F.bernal, Sep 21 '18 at 09:25

score 1 · Accepted Answer · answered Oct 08 '18 at 21:25

I believe this is because you are not using through2 correctly - i.e. not actually emptying the buffer once it's full (that's why it hangs at 10% on bigger files, but works on smaller ones).

I believe an implementation like this should do it:

const pass2 = through2(function(chunk, encoding, next) {

   // do something with the data


   // Use this only if you want to send the data further to another stream reader 
   // Note - From your implementation you don't seem to need it
   // this.push(data)

   // This is what tells through2 it's ready to empty the 
   //  buffer and read more data
   next();
})

That was the missing piece of logic. Thank you! – garajo Oct 10 '18 at 04:38 — garajo, Oct 10 '18 at 04:38

How to defer stream read invocation

1 Answers1