Feature Request: way to merge streams like pumpify(a, b)

[JasonWoof]

• Version: v13.6.0
• Platform: Linux 535297d0ce81 4.19.0-4-amd64 deps: update openssl to 1.0.1j #1 SMP Debian 4.19.28-2 (2019-03-15) x86_64 GNU/Linux
• Subsystem: stream

What steps will reproduce the bug?

const stream = require('stream');

// two different transforms (these work)
const transform_double = new stream.Transform({
    transform(chunk, encoding, callback) {
        this.push(chunk);
        this.push(chunk);
        callback();
    }
});
const transform_uppercase = new stream.Transform({
    transform(chunk, encoding, callback) {
        this.push(chunk.toString().toUpperCase());
        callback();
    }
});

// compose the transforms
const parser = stream.pipeline(transform_double, transform_uppercase, ()=>0);

// run stdin through it
process.stdin.pipe(parser).pipe(process.stdout);

Then pipe some text to it eg: echo "foo" | node reproduce.js

How often does it reproduce? Is there a required condition?

always

What is the expected behavior?

The text sent in should go through both transformations, eg output "FOOFOO"

What do you see instead?

The text piped in is only sent to the last transformation, eg output "FOO"

Additional information

The docs here don't say what this function returns:
https://nodejs.org/api/stream.html#stream_stream_pipeline_streams_callback

I'm saying that it should return a duplex stream so I can read and write to the ends of the pipeline.

[JasonWoof]

oh, that system/version is from me running it in the docker node:latest image. I get exactly the same results with v10.17.0 in debian unstable (not in docker).

[Trott]

@nodejs/streams

[vweevers]

@JasonWoof stream.pipeline() returns the last stream passed in, in the spirit of a.pipe(b).pipe(c). Which in your case is transform_uppercase.

To get a duplex stream instead, you might like pumpify.

[JasonWoof]

I don't see the point of stream.pipeline() if it does the same thing as a.pipe(b).pipe(c) and doesn't create something you can pipe data to.

Technically current stream.pipeline() does return something you can pipe to, but piping to it does something very weird (ie neither useful nor expected.)

I think node should have the ability to compose streams without pulling in god knows how many npm modules.

If stream.pipeline() worked the way I suggest, you could still do stream.pipeline(a,b).pipe(c) and you could also do a.pipe(stream.pipeline(b, c)).pipe(d)

[vweevers]

I don't see the point of stream.pipeline()

It forwards errors and does cleanup. That's cumbersome without stream.pipeline().

Technically current stream.pipeline() does return something you can pipe to, but piping to it does something very weird (ie neither useful nor expected.)

It returns the last stream (pipeline(a, b) === b). That's useful for composition, though not the way you expected it. I do agree the return value should be documented to avoid confusion.

If stream.pipeline() worked the way I suggest, you could still do stream.pipeline(a,b).pipe(c)

If stream.pipeline(a,b) were to return a duplex stream, that writes to the first stream and reads from the last, that's not quite the same. Such a utility is undoubtedly useful, which is why I suggested pumpify, but it must be separate from stream.pipeline().

[JasonWoof]

OK, sounds like stream.pipeline() isn't going to change, so I'll try making this a feature request for a built-in way to combine streams. I think it's good to have this sort of stuff built in because then more people will get the error forwarding, cleanup, and proper backpressure that they might otherwise not bother implementing (or even realize they need to).

[mcollina]

I would tend to agree that this would be a nice API to add. The main difficulty is to add it with very little perf overhead (no double buffering, like pumpify/duplexify do).

A good starting point could be to lift Duplexify and Pumpify and their tests and send a PR.

More than anything else, this feature needs somebody willing to devote time to it.

cc @ronag @mafintosh

[ronag]

A good starting point could be to lift Duplexify and Pumpify and their tests and send a PR.

I'm -1 on this. A lot of things has happened in streams and I'm not convinced the way these are implemented are the best way to achieve the discussed functionality.

More than anything else, this feature needs somebody willing to devote time to it.

I might be interested in this at some point in the future.

[JasonWoof]

I'm in learning mode with streams (ie hobby project no deadline) so I wrote my own pumpify as a learning exercise. I created a transform stream and did all the fiddling to pass along back-pressure and errors and whatnot.

But this could be much cleaner if it was built-in. All I want is a duplex stream interface that writes to the first stream in pipeline and reads from the last. I'm imagining something like:

function pipelineify (...streams) {
    // pipe to each other and forward errors
    stream.pipeline(...streams);
    // create a stream object that writes to the start
    // of the pipeline and reads from the end
    return new stream.Duplex({
        writeStream: streams[0],
        readStream: streams[streams.length - 1],
    });
}

[alesmenzel]

new Duplex({
  writeStream: <Writable>,
  readStream: <Readable>,
});

@JasonWoof from which version of Node is that Duplex writeStream/readStream options supported? I cannot find any mention of this in the docs.

• https://nodejs.org/docs/latest-v14.x/api/stream.html#stream_new_stream_duplex_options

@vweevers a question regarding .pipe(), it seems there is a lot of contradictory posts on whether I should handle the stream close/error/finish/end events, e.g. according to the Node docs it is not required to handle those events while piping, e.g. from this example

• https://nodejs.org/en/knowledge/advanced/streams/how-to-use-stream-pipe/
• https://nodejs.org/docs/latest-v14.x/api/stream.html#stream_new_stream_duplex_options

but then there are blog posts saying one should destroy the streams manually

• https://www.alxolr.com/articles/understanding-memory-leaks-in-node-js-part-1

what is the correct approach, should we handle destroying the streams when they close?

Actually there is a nearly hidden note on memory leaks due to not manually destroying the writable stream on read errors.

One important caveat is that if the Readable stream emits an error during processing, the Writable destination is not closed automatically. If an error occurs, it will be necessary to manually close each stream in order to prevent memory leaks.

So should we only care about the error event and if the stream ends without error does it destroy itself automatically?

For example, is this correctly handling the pipe (= only handling the readstream error events)?

const readable = fs.createReadStream('x.txt')
const writable = fs.createWriteStream('y.txt')

readable.once('error' , (err) => {
  writable.destroy(err)
})

readable.pipe(writable)

Sorry for a little off-topic questions.

[mcollina]

You should use https://nodejs.org/api/stream.html#stream_stream_pipeline_source_transforms_destination_callback or the pump module on npm

[JasonWoof]

@alesmenzel none to date. That is my feature request.

[alesmenzel]

@JasonWoof oh, I misread that