Promise-friendly stream.pipeline and stream.finished

I often use stream.pipeline in async functions, and it's always a bit painful to have to promisify it.
What about exporting a promisified version from the stream module directly?
I haven't opened an issue or PR before because I don't have a good idea for the name, but maybe somebody else does!

@nodejs/streams

When you do that, do you need to access the returned stream as well or is a Promise that gets settled when the callback version currently does?

What if streams exposed a (lazy) getter for the close event that returned a promise?

Such an API would be something like:

  await pipeline(fn1, fn2, fn3).closed;

Although the callback in pipeline would have to be optional in such a case which isn't too great.

We can also expose a stream.promises.pipelne that behaves exactly like the existing one but returns a Stream, promise object pair or something.

When you do that, do you need to access the returned stream as well or is a Promise that gets settled when the callback version currently does?

I've personally never used the return value, but it doesn't mean we shouldn't expose it somehow

When you do that, do you need to access the returned stream as well or is a Promise that gets settled when the callback version currently does?

That API should (IMO) not return the stream just a simple promise. If the user needs the the last stream he can simply keep a reference to it.

I would like to avoid trying to make this unnecessarily complex.

Another alternative is that we encourage the following pattern instead:

await stream.finished(stream.pipeline(fn1, fn2, fn3));

But then we have the problem that the callback argument is not optional :/. Maybe it should be?

@ronag

@ronag the problem is that it would be ambiguous and not backwards compatible right?

@ronag

@ronag the problem is that it would be ambiguous and not backwards compatible right?

I think it should be? What part do you think wouldn't be?

How do you safely disambiguate the callback from the "last (destination) stream" parameter? (given transforms can be "destinations" too)

I added stream.finished to the title :)

How do you safely disambiguate the callback from the "last (destination) stream" parameter? (given transforms can be "destinations" too)

Ah you mean the make the callback optional part. Not sure.

I would be fine with:

const ret = stream.pipeline(fn1, fn2, fn3, () => {});
await stream.finished(ret);

Even if it isn't optimal...

I would kind of like to deprecate the callback in favor for the above for the callback case as well. Not sure how we could do that though...

i.e.

const ret = stream.pipeline(fn1, fn2, fn3))
stream.finished(ret, (err) => { console.error(err) }));

vs

stream.pipeline(fn1, fn2, fn3, (err) => { console.error(err) }));

So a stream.promises.pipeline(source, [transforms ,] destination) that returns a promise for when the transformation has finished? (Takes no callback and returns no stream)

So a stream.promises.pipeline(source, [transforms ,] destination) that returns a promise for when the transformation has finished? (Takes no callback and returns no stream)

My suggestion is that pipeline always returns a stream and you use finished to "callbackify" and/or "promisify".

const ret = stream.pipeline(fn1, fn2, fn3);
await stream.finished(ret);

I would agree with this, however this would add some more event handler and overhead as we need to call finished in pipeline anyway. Maybe we could set a private symbol to avoid it.

Note that we did a significant back-and-forth on the pipeline callback. pump did not require one. Then we made node core require one. See #20437 for more details.

Note that we did a significant back-and-forth on the pipeline callback. pump did not require one. Then we made node core require one. See #20437 for more details.

I would suggest we revisit this. Looks like me that there were some concerns whether this was a good change or not. Maybe this could further shift the weight on the scale? I don't think changing it back would be a significant change.

however this would add some more event handler and overhead as we

If this is a concern there is work to be done in finished. We are already registering an unnecessary amount of handlers. We could probably remove some of them for the most common cases.

@ronag

+1, I agree with the direction @ronag

@ronag

+1, I agree with the direction @ronag

Looking into this further. @benjamingr point comes that it becomes ambigious to determine whether the last function is a transform function or a callback. So making the callback optional might not actually be possible since we added the async generator function support.

Another approach would be to create a new function e.g. pump, pipeline2 (:smile:) which is different from pipeline on two points:

no callback
if the first stream is a Duplex, it returns a new Duplex which will write to the first stream and read from the last stream. For better composition.

Then we would leave pipeline as is and possible deprecate it in the future?

I've already done some work on this direction in the past and would not mind making a PR if it sounds viable.

I've already done some work on this direction in the past and would not mind making a PR if it sounds viable.

If another function is added, we can just call it "the promise variant" and make that return a promise (and not the stream) like you suggested here.

Then I would prefer to not have to type two commands (one to compose the streams and one to know when it is finished) - I would prefer to just do:

// stuff before pipeline
await pipeline(fn1, fn2, fn3);
// stuff after pipeline

If another function is added, we can just call it "the promise variant" and make that return a promise (and not the stream) like you suggested here.

Would like to point out that we already have a feature request (which I find reasonable) for adding another function that does:

if the first stream is a Duplex, it returns a new Duplex which will write to the first stream and read from the last stream. For better composition.

i.e. we might have to add another function with a different name anyway that might supersede pipeline.

we can just call it "the promise variant" and make that return a promise (and not the stream)

I'm good with that as well. I guess one does not exclude the other.

if the first stream is a Duplex, it returns a new Duplex which will write to the first stream and read from the last stream. For better composition.

This should be done only if needed. It should not be a generic function.

I think we can add as many variant as we need.

stream: add promises version to utility functions

Navigation Menu

Repository navigation

Promise-friendly stream.pipeline and stream.finished #33582

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Promise-friendly stream.pipeline and stream.finished #33582

Description

Activity

benjamingr commented on May 27, 2020

targos commented on May 27, 2020

ronag commented on May 27, 2020

ronag commented on May 27, 2020

benjamingr commented on May 27, 2020

ronag commented on May 27, 2020

benjamingr commented on May 27, 2020

targos commented on May 27, 2020

ronag commented on May 27, 2020

benjamingr commented on May 27, 2020

ronag commented on May 27, 2020

mcollina commented on May 27, 2020

ronag commented on May 27, 2020

ronag commented on May 27, 2020

mcollina commented on May 27, 2020

ronag commented on May 27, 2020

benjamingr commented on May 27, 2020

ronag commented on May 27, 2020

ronag commented on May 27, 2020

mcollina commented on May 27, 2020

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions