stream: fix legacy pipe error handling

Fixes: #35237 PR-URL: #35257 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Luigi Pinca <luigipinca@gmail.com> Reviewed-By: Rich Trott <rtrott@gmail.com>

Author: ronag

lib/internal/streams/legacy.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const {
+ ArrayIsArray,
  ObjectSetPrototypeOf,
 } = primordials;
 
@@ -58,12 +59,12 @@ Stream.prototype.pipe = function(dest, options) {
  function onerror(er) {
  cleanup();
  if (EE.listenerCount(this, 'error') === 0) {
- throw er; // Unhandled stream error in pipe.
+ this.emit('error', er);
  }
  }
 
- source.on('error', onerror);
- dest.on('error', onerror);
+ prependListener(source, 'error', onerror);
+ prependListener(dest, 'error', onerror);
 
  // Remove all the event listeners that were added.
  function cleanup() {
@@ -92,4 +93,22 @@ Stream.prototype.pipe = function(dest, options) {
  return dest;
 };
 
-module.exports = Stream;
+function prependListener(emitter, event, fn) {
+ // Sadly this is not cacheable as some libraries bundle their own
+ // event emitter implementation with them.
+ if (typeof emitter.prependListener === 'function')
+ return emitter.prependListener(event, fn);
+
+ // This is a hack to make sure that our error handler is attached before any
+ // userland ones. NEVER DO THIS. This is here only because this code needs
+ // to continue to work with older versions of Node.js that do not include
+ // the prependListener() method. The goal is to eventually remove this hack.
+ if (!emitter._events || !emitter._events[event])
+ emitter.on(event, fn);
+ else if (ArrayIsArray(emitter._events[event]))
+ emitter._events[event].unshift(fn);
+ else
+ emitter._events[event] = [fn, emitter._events[event]];
+}
+
+module.exports = { Stream, prependListener };

lib/internal/streams/readable.js

@@ -22,7 +22,6 @@
 'use strict';
 
 const {
- ArrayIsArray,
  NumberIsInteger,
  NumberIsNaN,
  NumberParseInt,
@@ -38,7 +37,7 @@ module.exports = Readable;
 Readable.ReadableState = ReadableState;
 
 const EE = require('events');
-const Stream = require('internal/streams/legacy');
+const { Stream, prependListener } = require('internal/streams/legacy');
 const { Buffer } = require('buffer');
 
 let debug = require('internal/util/debuglog').debuglog('stream', (fn) => {
@@ -69,24 +68,6 @@ function nop() {}
 
 const { errorOrDestroy } = destroyImpl;
 
-function prependListener(emitter, event, fn) {
- // Sadly this is not cacheable as some libraries bundle their own
- // event emitter implementation with them.
- if (typeof emitter.prependListener === 'function')
- return emitter.prependListener(event, fn);
-
- // This is a hack to make sure that our error handler is attached before any
- // userland ones. NEVER DO THIS. This is here only because this code needs
- // to continue to work with older versions of Node.js that do not include
- // the prependListener() method. The goal is to eventually remove this hack.
- if (!emitter._events || !emitter._events[event])
- emitter.on(event, fn);
- else if (ArrayIsArray(emitter._events[event]))
- emitter._events[event].unshift(fn);
- else
- emitter._events[event] = [fn, emitter._events[event]];
-}
-
 function ReadableState(options, stream, isDuplex) {
  // Duplex streams are both readable and writable, but share
  // the same options object.

lib/internal/streams/writable.js

@@ -38,7 +38,7 @@ module.exports = Writable;
 Writable.WritableState = WritableState;
 
 const EE = require('events');
-const Stream = require('internal/streams/legacy');
+const Stream = require('internal/streams/legacy').Stream;
 const { Buffer } = require('buffer');
 const destroyImpl = require('internal/streams/destroy');
 const {

lib/stream.js

@@ -25,9 +25,7 @@ const pipeline = require('internal/streams/pipeline');
 const eos = require('internal/streams/end-of-stream');
 const internalBuffer = require('internal/buffer');
 
-// Note: export Stream before Readable/Writable/Duplex/...
-// to avoid a cross-reference(require) issues
-const Stream = module.exports = require('internal/streams/legacy');
+const Stream = module.exports = require('internal/streams/legacy').Stream;
 Stream.Readable = require('internal/streams/readable');
 Stream.Writable = require('internal/streams/writable');
 Stream.Duplex = require('internal/streams/duplex');

test/parallel/test-stream-pipe-error-handling.js

@@ -22,7 +22,7 @@
 'use strict';
 const common = require('../common');
 const assert = require('assert');
-const Stream = require('stream').Stream;
+const { Stream, PassThrough } = require('stream');
 
 {
  const source = new Stream();
@@ -108,3 +108,17 @@ const Stream = require('stream').Stream;
  w.removeListener('error', () => {});
  removed = true;
 }
+
+{
+ const _err = new Error('this should be handled');
+ const destination = new PassThrough();
+ destination.once('error', common.mustCall((err) => {
+ assert.strictEqual(err, _err);
+ }));
+
+ const stream = new Stream();
+ stream
+ .pipe(destination);
+
+ destination.destroy(_err);
+}