stream: add stream.pipeline and stream.onEnd #19828

mafintosh · 2018-04-05T10:43:10Z

Checklist

make -j4 test (UNIX), or vcbuild test (Windows) passes
tests and/or benchmarks are included
documentation is changed or added
commit message follows commit guidelines

Adds stream.pipeline(...streams, [cb]) and stream.onEnd(stream, cb) based on my two modules https://github.com/mafintosh/pump and https://github.com/mafintosh/end-of-stream.

Needs tests+docs (writing those now), but thought it'd be good to get this up now for feedback.

Supersedes #13506, as there's been changes since so making a new PR was easier

mafintosh · 2018-04-05T11:37:13Z

Goal is to get this out in 10, backports might be tricky as the core streams in 9, doesn't play well with error handling patterns used by pump.

mscdex · 2018-04-05T12:07:56Z

Isn't the second commit (changing the order of emitted stream events) unrelated and would be better as a separate PR?

mscdex · 2018-04-05T12:10:04Z

lib/stream.js

@@ -33,6 +35,9 @@ Stream.Duplex = require('_stream_duplex');
 Stream.Transform = require('_stream_transform');
 Stream.PassThrough = require('_stream_passthrough');

+Stream.pipeline = pump;
+Stream.onEnd = eos;


onEnd is a confusing name in the context of streams, we should use something else

Open for suggestions :)

Can we used finished, like in mississippi?

mscdex · 2018-04-05T12:12:42Z

I think the two new internal modules could use some improvements code style-wise (they may not even pass the linter?) and performance-wise (e.g. replacing functional array methods).

mscdex · 2018-04-05T12:14:36Z

Also, why does this need to live on require('stream') and not the more appropriate and more specific stream constructor(s)?

mafintosh · 2018-04-05T12:30:40Z

@mscdex the order change is to make the streams play well with the pump code, and how it, in general should be emitted (mistake on my part when I updated the error handling). Happy to move that out if that's easier to review.

I'd very much prefer not to rewrite the source at this point. This is ported with minimal changes from my pump and end-of-stream module, to pass the node core linter, and there is lots of complexity in there.

I'm not I understand what you mean by constructors?

mscdex · 2018-04-05T13:25:09Z

I'd very much prefer not to rewrite the source at this point. This is ported with minimal changes from my pump and end-of-stream module, to pass the node core linter, and there is lots of complexity in there.

IMO if it's being included directly in node core, we should strive to make the code as readable as possible and performant as possible (especially in this case because the public functions being added could be in hot paths). There isn't even that much code there, so I don't see why this is a problem. Some of the changes would even be pretty straightforward (e.g. replacing the functional array methods).

I'm not I understand what you mean by constructors?

e.g. require('stream').Readable

mcollina

Can you add some test regarding to http.OutgoingMessage (both on a client and on a server)? They are the special writable-like object.

Also for HTTP2 compat API.

mcollina · 2018-04-05T13:26:19Z

lib/stream.js

@@ -22,6 +22,8 @@
 'use strict';

 const { Buffer } = require('buffer');
+const pump = require('internal/streams/pump');


I would name this pipeline instead.

mcollina · 2018-04-05T13:27:41Z

test/parallel/test-stream-pipeline.js

+    });
+  }
+
+  run();


you need to call common.crashOnUnhandledRejection() at the top of the file.

mcollina · 2018-04-05T13:28:05Z

test/parallel/test-stream-pipeline.js

+
+    await pipelinePromise(read, write);
+
+    // nexttick to avoid the promise catching it


this nextTick is not clear to me.

was to work around the need for common.crashOnUnhandledRejection() which i didn't know about 👍

mcollina · 2018-04-05T13:28:23Z

test/parallel/test-stream-pipeline.js

+  read.push('data');
+  setImmediate(() => read.destroy(new Error('kaboom')));
+
+  pipeline(read, write, common.mustCall((err) => {


can you add some test for the return value of pipeline?

mcollina · 2018-04-05T13:29:01Z

lib/internal/streams/pump.js

+  if (Array.isArray(streams[0])) streams = streams[0];
+
+  if (streams.length < 2) {
+    throw new ERR_ASSERTION('pipeline requires two streams per minimum');


I don't think this is the right error.

It's an assertion error no?

mcollina · 2018-04-05T13:30:36Z

test/parallel/test-http2-stream-destroy-event-order.js

-  req.on('close', common.mustCall(() => {
-    req.on('error', common.mustCall(() => {
+  req.on('error', common.mustCall(() => {
+    req.on('close', common.mustCall(() => {


are these changes part of this PR, if so why?

I messed up the emit order in my streams error handling PR (error should be before close). Otherwise these modules will report a premature close error instead of the error passed to stream.destroy. It's quite related to the end-of-stream/pump stuff so just bundled that in here. Want me to open an independent PR with that? (tests won't pass without this fix)

yes please! If possible I would like to backport this.

👍 I don't think a backport is needed as the stream error handling stuff we did is going out in 10 only I think?

I mean, I would like to backport this PR to 8 if possible.

I'm splitting it out in a sec

mafintosh · 2018-04-05T13:32:21Z

@mscdex Still not sure I follow, you want this exposed as stream.Readable.pipeline?

jasnell · 2018-04-05T14:11:20Z

lib/internal/streams/pipeline.js

@@ -0,0 +1,95 @@
+// Ported from https://github.com/mafintosh/pump with
+// permission from the author, Mathias Buus (@mafintosh).


if the original code has a copyright/license, then it should likely either be included here or embedded in the LICENSE file.

It does, but I did mention here I'm fine waiving copyright, mafintosh/pump#17 (comment)

(I'm fine with whatever is easiest btw)

mcollina · 2018-04-05T14:14:44Z

lib/internal/streams/pipeline.js

+  if (Array.isArray(streams[0])) streams = streams[0];
+
+  if (streams.length < 2) {
+    throw new ERR_ASSERTION('pipeline requires two streams per minimum');


I think that is for the assert module.

You should probably be using ERR_MISSING_ARGS here.

mcollina · 2018-04-05T14:23:15Z

lib/internal/streams/pipeline.js

+  return streams.pop();
+}
+
+function pump(...streams) {


can you name this pipeline?

jasnell · 2018-04-05T17:15:25Z

test/parallel/test-stream-pipeline.js

+      // where the client keeps the event loop going
+      // (replacing the rs.destroy() with req.end()
+      // exits it so seems to be a destroy bug there
+      client.unref();


will be investigating this later on this afternoon :-)

Testing this with other Readable instances shows this working. For instance, replace the custom Readable above with const rs = fs.createReadStream(__filename) and the test appears to work just fine.

Yep, just verified, commenting out the client.unref() and replacing the custom Readable with fs.createReadStream() and adjusting the cnt below a bit to account for the difference in the number of data events allows this test to pass and the Http2Stream shuts down normally. So the issue appears to be specific to the custom Readable?? /cc @mcollina

@jasnell hmm, i don't understand how the Readable should affect this. it's just the trigger of the destruction of the pipeline. could it be that there is an http2 issue if a bunch of writes happen while the stream is destroyed?

Not that I'm aware of. At this point I'm a bit unsure what could be happening. I've tried a number of variations but can only reproduce it with the custom Readable. I've asked @mcollina to take a look. He said he will take a look tomorrow.

@mcollina... Please don't forget to take a look at this one :)

mafintosh · 2018-04-06T11:10:50Z

@mcollina PTAL (docs missing, working on that now)

vsemozhetbyt

Sorry for many petty doc nits)

vsemozhetbyt · 2018-04-06T20:40:24Z

doc/api/errors.md

+<a id="ERR_STREAM_PREMATURE_CLOSE"></a>
+### ERR_STREAM_PREMATURE_CLOSE
+
+An error returned by `stream.onEnd` and `stream.pipeline`, when a stream


Nit: `stream.onEnd` and `stream.pipeline` -> `stream.onEnd()` and `stream.pipeline()`?

It should be stream.finished() and not stream.onEnd(), since the former is what is currently used in this PR.

vsemozhetbyt · 2018-04-06T20:43:30Z

doc/api/stream.md

@@ -46,6 +46,8 @@ There are four fundamental stream types within Node.js:
 * [Transform][] - Duplex streams that can modify or transform the data as it
  is written and read (for example [`zlib.createDeflate()`][]).

+Additionally this module includes the utility functions [pipeline][] and [finished][].


This line should be wrapped after 80 characters.

vsemozhetbyt · 2018-04-06T20:56:31Z

doc/api/stream.md

@@ -1263,6 +1265,106 @@ implementors should not override this method, but instead implement
 [`readable._destroy`][readable-_destroy].
 The default implementation of `_destroy` for `Transform` also emit `'close'`.

+### stream.pipeline(...streams[, callback])


Is it OK to use rest parameters syntax with non-last parameter even just for convenience?

It seems stream.pipeline() section should go after stream.finished() section, alphabetically.

vsemozhetbyt · 2018-04-06T21:01:54Z

doc/api/stream.md

+* `callback` {Function} A callback function that takes an optional error
+  argument.
+
+A class method to pipe between streams forwarding errors and properly cleaning


class method -> module method?

vsemozhetbyt · 2018-04-06T21:04:41Z

doc/api/stream.md

+const fs = require('fs');
+const zlib = require('zlib');
+
+// Use pipeline api to easily pipe a series of streams


api -> API?

vsemozhetbyt · 2018-04-06T21:16:36Z

doc/api/stream.md

+  console.log('Pipeline succeded');
+}
+
+run().catch(console.log);


.catch(console.log) -> .catch(console.error)?

vsemozhetbyt · 2018-04-06T21:19:22Z

doc/api/stream.md

+
+finished(rs, (err) => {
+  if (err) {
+    console.log('Stream failed', err);


console.log() -> console.error()?

vsemozhetbyt · 2018-04-06T21:25:28Z

doc/api/stream.md

+
+Especially useful in error handling scenarios where a stream is destroyed
+prematurely (like an aborted HTTP request), and will not emit `end`
+or `finish`.


Not sure, but maybe `end` or `finish` -> `'end'` or `'finish'` events?

vsemozhetbyt · 2018-04-06T21:25:55Z

doc/api/stream.md

+prematurely (like an aborted HTTP request), and will not emit `end`
+or `finish`.
+
+The `finished` API is promisify'able as well


as well -> as well:?

vsemozhetbyt · 2018-04-06T21:27:44Z

doc/api/stream.md

+  console.log('Stream is done reading');
+}
+
+run().catch(console.log);


.catch(console.log) -> .catch(console.error)?

mafintosh · 2018-04-09T12:49:16Z

@vsemozhetbyt fixed your doc nits

mafintosh · 2018-04-09T12:58:01Z

Rebased as #19836 landed, PTAL

mcollina

still LGTM

mafintosh · 2018-04-09T19:07:30Z

CI: https://ci.nodejs.org/job/node-test-pull-request/14149/

ronkorving · 2018-04-10T01:24:33Z

@mafintosh Sorry for bikeshedding. I'm not a native English speaker, but would like to ask... Can pipeline reasonably be used as a verb? If so, LGTM 👍 If not, I personally prefer a verby method, rather than a noun (which I read pipeline as). This could just be my lack of knowledge of the English language though.

mafintosh · 2018-04-10T11:18:03Z

@ronkorving Not a native speaker as well, but I use pipeline as a verb and I think that's correct.

mafintosh · 2018-04-11T12:33:49Z

@mscdex I plan on landing this Monday, so PTAL before if you have time :)

PR-URL: #19828 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>

mafintosh · 2018-04-16T14:05:22Z

Landed in f64bebf

PR-URL: #19828 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>

PR-URL: nodejs#19828 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>

nodejs-github-bot added the lib / src Issues and PRs related to general changes in the lib or src directory. label Apr 5, 2018

vsemozhetbyt added stream Issues and PRs related to the stream subsystem. semver-minor PRs that contain new features and should be released in the next minor version. labels Apr 5, 2018

mafintosh force-pushed the add-pump-and-eos branch 3 times, most recently from 6cc85d2 to ca4f736 Compare April 5, 2018 11:34

mafintosh force-pushed the add-pump-and-eos branch from ca4f736 to 7b5e84b Compare April 5, 2018 12:01

mscdex reviewed Apr 5, 2018

View reviewed changes

mafintosh force-pushed the add-pump-and-eos branch from 7b5e84b to 1593e58 Compare April 5, 2018 12:15

mafintosh force-pushed the add-pump-and-eos branch from 1593e58 to f38fd9b Compare April 5, 2018 12:33

mcollina reviewed Apr 5, 2018

View reviewed changes

mafintosh force-pushed the add-pump-and-eos branch 3 times, most recently from c128bad to bd162fa Compare April 5, 2018 13:51

jasnell reviewed Apr 5, 2018

View reviewed changes

mcollina reviewed Apr 5, 2018

View reviewed changes

lib/internal/streams/pipeline.js Outdated

return streams.pop();

}

function pump(...streams) {

Copy link

Member

mcollina Apr 5, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you name this pipeline?

mafintosh force-pushed the add-pump-and-eos branch from bd162fa to 4f83f94 Compare April 5, 2018 17:11

jasnell reviewed Apr 5, 2018

View reviewed changes

mafintosh mentioned this pull request Apr 5, 2018

stream: always emit error before close #19836

Closed

4 tasks

mafintosh force-pushed the add-pump-and-eos branch from 4f83f94 to 3a117e5 Compare April 6, 2018 11:09

mafintosh mentioned this pull request Apr 6, 2018

Allow to await writable streams #19068

Closed

mafintosh mentioned this pull request Apr 6, 2018

http2: add failing writes with destroy test #19852

Closed

4 tasks

vsemozhetbyt reviewed Apr 6, 2018

View reviewed changes

vsemozhetbyt added the doc Issues and PRs related to the documentations. label Apr 6, 2018

mafintosh force-pushed the add-pump-and-eos branch from 59b3fea to 7a4afd3 Compare April 9, 2018 11:49

mafintosh added 5 commits April 9, 2018 14:49

stream: add pipeline and onEnd

b636f8b

stream: add pipeline test

2346ca1

stream: add tests for onEnd

61ef762

stream: onEnd -> finished

c66216d

stream: add docs

89523b2

mafintosh force-pushed the add-pump-and-eos branch from 7a4afd3 to 89523b2 Compare April 9, 2018 12:56

mcollina approved these changes Apr 9, 2018

View reviewed changes

jasnell added this to the 10.0.0 milestone Apr 9, 2018

BridgeAR mentioned this pull request Apr 10, 2018

Stream pump #13506

Closed

jasnell approved these changes Apr 10, 2018

View reviewed changes

mafintosh added a commit that referenced this pull request Apr 16, 2018

stream: add pipeline and finished

f64bebf

PR-URL: #19828 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>

mafintosh closed this Apr 16, 2018

mafintosh deleted the add-pump-and-eos branch April 16, 2018 14:05

jasnell pushed a commit that referenced this pull request Apr 16, 2018

stream: add pipeline and finished

a5cf3fe

PR-URL: #19828 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>

BridgeAR pushed a commit to BridgeAR/node that referenced this pull request May 1, 2018

stream: add pipeline and finished

723cfa2

PR-URL: nodejs#19828 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>

Trott mentioned this pull request Nov 25, 2018

meta: patch branches #24467

Closed

lukesims mentioned this pull request Jun 5, 2019

Current project status mafintosh/pump#46

Closed

This was referenced Nov 28, 2022

--- imcotton/npm-installer#1

Closed

set 1s delay for untar pipes to avoid spawn racing purescript/npm-installer#38

Closed

+                  });
+                }
+                run();


		await pipelinePromise(read, write);

		// nexttick to avoid the promise catching it

		@@ -0,0 +1,95 @@
		// Ported from https://github.com/mafintosh/pump with
		// permission from the author, Mathias Buus (@mafintosh).

stream: add stream.pipeline and stream.onEnd #19828

stream: add stream.pipeline and stream.onEnd #19828

Conversation

mafintosh commented Apr 5, 2018 • edited Loading

Checklist

mafintosh commented Apr 5, 2018

mscdex commented Apr 5, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mscdex commented Apr 5, 2018 • edited Loading

mscdex commented Apr 5, 2018

mafintosh commented Apr 5, 2018

mscdex commented Apr 5, 2018

mcollina left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mafintosh commented Apr 5, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mafintosh commented Apr 6, 2018

vsemozhetbyt left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mafintosh commented Apr 9, 2018

mafintosh commented Apr 9, 2018

mcollina left a comment

Choose a reason for hiding this comment

mafintosh commented Apr 9, 2018

ronkorving commented Apr 10, 2018

mafintosh commented Apr 10, 2018

mafintosh commented Apr 11, 2018

mafintosh commented Apr 16, 2018

mafintosh commented Apr 5, 2018 •

edited

Loading

mscdex commented Apr 5, 2018 •

edited

Loading