Pipe event conflict with readable-stream v3 / node.js core

[coreyfarrell]

readable-stream v3 and node.js core use dest.emit('pipe', src) while streamx uses src.emit('pipe', dest).

Demonstration script:

'use strict';

const streamx = require('streamx');
const stream = require('readable-stream');

const ids = new Map();
function watchPipe(self, stream) {
	ids.set(stream, self);
	stream.on('pipe', other => {
		console.log('pipe', {
			self,
			other: ids.get(other)
		});
	});
}

function runTest(id, stream) {
	watchPipe('stdout', process.stdout);
	watchPipe('stdin', process.stdin);
	watchPipe(id, stream);

	stream.pipe(process.stdout);
	process.stdin.pipe(stream);
}

if (process.argv.slice(2)[0]) {
	runTest('streamx', new streamx.Transform());
} else {
	runTest('readable-stream', new stream.PassThrough());
}

Run the script with and without an argument:

$ echo | node t.js
pipe { self: 'stdout', other: 'readable-stream' }
pipe { self: 'readable-stream', other: 'stdin' }

$ echo | node t.js 1
pipe { self: 'streamx', other: 'stdout' }
pipe { self: 'streamx', other: 'stdin' }

[mafintosh]

Ah didn't know! Thats a bug. Will fix.

[coreyfarrell]

Additional somewhat related comment streamx.pipe(dest) does not add a data listener to streamx as stream v3 does. My goal is to migrate to-through from though2 to streamx.Transform. The following changes to the end of ReadableState#pipe make this switch:

+    this.stream.once('data', noop)
+    pipeTo.emit('pipe', this.stream)
-    this.stream.emit('pipe', pipeTo)

Adding the listener for data causes this.stream.emit('newListener', 'data') which is a signal to start the flow. Technically stream v3 uses this.stream.on('data', ...) keeping the data listener active, I'm not sure if anything actually looks for the data listener to remain active. The code I've seen gets triggered by newListener events for data or readable.

I cloned streamx and tested the above patch, it didn't cause any new failures. On node.js 12+ I received a failure from test/compat.js on nodeStream.finished writable before and after making the above change, tests still passed on node.js 10. I assume this is due to nodejs/node#29699.

I can open a separate issue for adding a data listener if you prefer?

[phated]

@coreyfarrell thanks for working on these things!

@mafintosh heads up that streamx is going to be used inside gulp v5, which we are pushing on right now. 🎉 Thanks for all the great work!

[mafintosh]

@coreyfarrell let's do another issue for the data listener, cause it triggers slow paths in streamx adding one, so let's see if we can't solve it

@phated uhhhh!

[phated]

Something that I thought would be worth mentioning is that I've often wanted src.on('pipe'), so maybe streamx can emit on both?

[coreyfarrell]

Something that I thought would be worth mentioning is that I've often wanted src.on('pipe'), so maybe streamx can emit on both?

It is important that we be able to know the direction of the pipe, I think it would be problematic if stream.on('pipe') got events when stream is source and destination.

[mafintosh]

What about:

dst.on('pipe')
src.on('piping')

Then we have compat and an event on both ends (someone better at naming than me chime in :D)

[coreyfarrell]

@phated do you think a separate event specifically for 'piping out' is useful or is knowing something wants data from src what we really need? See #16 and nodejs/node#33567.

[phated]

@coreyfarrell I'm not sure of the question exactly, but if you are asking which events the to-through module needs: it currently looks at the "wrapped" stream being piped "into" (which disables the automatic flow of the underlying readable) and when the stream they were piped to starts requesting data (so they can read data and pass it along). Though, I'm not sure if knowing that a stream was piped "from" would be enough to start reading data and pushing it into the buffer (it might be!).

[phated]

Btw, I think pipe for compat and piping are actually pretty good names!

[phated]

I just saw this from Matteo:

Specifically because we should not just check for pipes, but if the stream is being actively read from or not. One can consume a stream either 'data' or 'readable', which are used by 'pipe' and async iteration respectively.

I think that the to-through module would be reduced to only a few lines if there were an event to know when data was being read from the stream. A lot of the boilerplate in that module was setting up the listeners to emulate that.

[coreyfarrell]

Yes this is why I asked. I don't know of a situation where you'd need to know "this is specifically being piped out", generally you need to know "something wants data from this stream". This is the case in to-through.

[phated]

A good case for "this is specifically being piped out" is to emit extra metadata when a stream is piped (e.g. when Browserify is piped, they could emit the filename to the upstream consumer)

^ The important part is that the emit would contain a reference to the stream so you could emit more data to it.

[phated]

Where'd we land on getting this changed?

[mafintosh]

I’ll land this today :)

[mafintosh]

Fixed and released in latest patch (bug fix).

I've also documented pipe/piping now so they are covered by semver :)

[coreyfarrell]

I just did some testing and I think this issue is resolved. Thanks @mafintosh!