Repository navigation
Promise-friendly stream.pipeline and stream.finished #33582
Description
Activity
When you do that, do you need to access the returned stream as well or is a
Promisethat gets settled when the callback version currently does?What if streams exposed a (lazy) getter for the
closeevent 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.pipelnethat behaves exactly like the existing one but returns aStream, promiseobject 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?
Reacted by Benjamin Gruenbaum@ronag the problem is that it would be ambiguous and not backwards compatible right?
How do you safely disambiguate the callback from the "last (destination) stream" parameter? (given transforms can be "destinations" too)
- changed the title
[-]Promise-friendly stream.pipeline[/-][+]Promise-friendly stream.pipeline and stream.finished[/+]on May 27, 2020 I added
stream.finishedto 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
finishedto "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
finishedin pipeline anyway. Maybe we could set a private symbol to avoid it.Note that we did a significant back-and-forth on the
pipelinecallback.pumpdid 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.+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 frompipelineon two points:- no
callback - if the first stream is a
Duplex, it returns a newDuplexwhich will write to the first stream and read from the last stream. For better composition.
Then we would leave
pipelineas 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.
- no
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.
- added a commit that references this issue
on Jul 27, 2026
I often use
stream.pipelinein async functions, and it's always a bit painful to have to promisify it.What about exporting a promisified version from the
streammodule 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