Make stream internal methods support promises / async await

[lmammino]

Hello all, It would be great if streams internal methods such as _write, _transform would support also a promise based implementation.

To clarify what I mean, here's an example.

Classic implementation of a writable streams that receives objects containing a path and a content and it will write a file with such content for every received chunk.

Classic implementation

import { Writable } from 'stream'
import { promises as fs } from 'fs'
import { dirname } from 'path'
import mkdirp from 'mkdirp'

export class ToFileStream extends Writable {
  constructor (opts) {
    super({ ...opts, objectMode: true })
  }

  async _write (chunk, encoding, cb) {
    mkdirp(dirname(chunk.path), (err) => {
      if (err) {
        return cb(err)
      }

      fs.writeFile(chunk.path, chunk.content, cb)
    })
  }
}

Nice and standard, but as it happens with callback-based APIs, it's easy to end up with a callback hell situation and repeated code in case of conditional asynchronous work.

Desired implementation

This does not work as of today

import { Writable } from 'stream'
import { promises as fs } from 'fs'
import { dirname } from 'path'
import mkdirp from 'mkdirp-promise'

export class ToFileStream extends Writable {
  constructor (opts) {
    super({ ...opts, objectMode: true })
  }

  async _write (chunk, encoding) {
    await mkdirp(dirname(chunk.path))
    return fs.writeFile(chunk.path, chunk.content)
  }
}

This would feel quite clean and standard for those used to the async await style

What we can do today

import { Writable } from 'stream'
import { promises as fs } from 'fs'
import { dirname } from 'path'
import mkdirp from 'mkdirp-promise'

export class ToFileStream extends Writable {
  constructor (opts) {
    super({ ...opts, objectMode: true })
  }

  async _write (chunk, encoding, cb) {
    try {
      await mkdirp(dirname(chunk.path))
      await fs.writeFile(chunk.path, chunk.content)
      cb()
    } catch (err) {
      cb(err)
    }
  }
}

In a situation with a lot of callbacks this might help to remove the nesting, but it still feels weird to (and hard to read) to mix async await style with callbacks.

---

What do you think? Would this be something feasible without breaking existing streams implementations?

[ronag]

I think it's feasible but not sure of the performance implications.

[lmammino]

Hello @ronag and thanks for the feedback.

Do you expect that async will add a delay or is it the check for different signatures that might add some significant overhead?

[ronag]

check for different signatures that might add some significant overhead?

this

[ronag]

Doesn't seem too bad actually. Though I'm not sure whether this is worth it.

We recently merged https://github.com/nodejs/node/pull/31223 which allows you to use generators which actually do what you are asking for here even more elegantly.

So you could do:

async function toFileStream(source) {
  for await (const chunk of source) {
      await mkdirp(dirname(chunk.path))
      await fs.writeFile(chunk.path, chunk.content)
  }
}

await pipeline(src, toFileStream)

[lmammino]

Thanks for your example @ronag, definitely useful and interesting.

Doesn't seem too bad actually. Though I'm not sure whether this is worth it.

My rationale is only from a syntactic perspective. I think it might be a bit confusing if you want to use async from one of these methods to have to call a callback at the end. I think it could be a nicer addition to the stream functionality to support this and it could also make them a bit more friendly to beginners.

How would you benchmark streams? Are there already tools in place in this repo (or others)?

[ronag]

How would you benchmark streams? Are there already tools in place in this repo (or others)?

node benchmark/compare.js --new ./out/Release/node --old ../node-master/out/Release/node --filter writable-manywrites -- streams | Rscript benchmark/compare.R

[ChocolateLoverRaj]

@lmammino I also support this idea. One potential problem though is that there would be different arguments for normal functions and async functions. For example, what if someone has something like this:

import { Writable } from 'stream'

export class SomeWritable extends Writable {
  async _write (chunk, encoding, callback) {
    callback("Callbacking");
   return "and returning a promise";
}

That would mess things up because the _write function could callback() something and also return a Promise.

@ronag Maybe there could be something like

import { Writable } from 'stream/promises'

Similar to

import { writeFile } from 'fs/promises'

[github-actions[bot]]

There has been no activity on this feature request for 5 months and it is unlikely to be implemented. It will be closed 6 months after the last non-automated comment.

For more information on how the project manages feature requests, please consult the feature request management document.

[github-actions[bot]]

There has been no activity on this feature request and it is being closed. If you feel closing this issue is not the right thing to do, please leave a comment.

For more information on how the project manages feature requests, please consult the feature request management document.