Updated to node v7.10.0.

README.md

@@ -18,7 +18,7 @@ npm install --save readable-stream
 This package is a mirror of the Streams2 and Streams3 implementations in
 Node-core.
 
-Full documentation may be found on the [Node.js website](https://nodejs.org/dist/v7.8.0/docs/api/stream.html).
+Full documentation may be found on the [Node.js website](https://nodejs.org/dist/v7.10.0/docs/api/stream.html).
 
 If you want to guarantee a stable streams base, regardless of what version of
 Node you, or the users of your libraries are using, use **readable-stream** *only* and avoid the *"stream"* module in Node-core, for background see [this blogpost](http://r.va.gg/2014/06/why-i-dont-use-nodes-core-stream-module.html).

lib/_stream_readable.js

@@ -482,7 +482,7 @@ Readable.prototype.pipe = function (dest, pipeOpts) {
 
   var doEnd = (!pipeOpts || pipeOpts.end !== false) && dest !== process.stdout && dest !== process.stderr;
 
-  var endFn = doEnd ? onend : cleanup;
+  var endFn = doEnd ? onend : unpipe;
   if (state.endEmitted) processNextTick(endFn);else src.once('end', endFn);
 
   dest.on('unpipe', onunpipe);
@@ -515,7 +515,7 @@ Readable.prototype.pipe = function (dest, pipeOpts) {
     dest.removeListener('error', onerror);
     dest.removeListener('unpipe', onunpipe);
     src.removeListener('end', onend);
-    src.removeListener('end', cleanup);
+    src.removeListener('end', unpipe);
     src.removeListener('data', ondata);
 
     cleanedUp = true;

test/common.js

@@ -377,6 +377,7 @@ if (global.__coverage__) knownGlobals.push(__coverage__);
 function leakedGlobals() {
   var leaked = [];
 
+  // eslint-disable-next-line no-var
   for (var val in global) {
     if (!knownGlobals.includes(global[val])) leaked.push(val);
   }if (global.__coverage__) {
@@ -564,15 +565,46 @@ exports.isAlive = function isAlive(pid) {
   }
 };
 
-exports.expectWarning = function (name, expected) {
-  if (typeof expected === 'string') expected = [expected];
-  process.on('warning', exports.mustCall(function (warning) {
+function expectWarning(name, expectedMessages) {
+  return exports.mustCall(function (warning) {
     assert.strictEqual(warning.name, name);
-    assert.ok(expected.includes(warning.message), 'unexpected error message: "' + warning.message + '"');
+    assert.ok(expectedMessages.includes(warning.message), 'unexpected error message: "' + warning.message + '"');
     // Remove a warning message after it is seen so that we guarantee that we
     // get each message only once.
-    expected.splice(expected.indexOf(warning.message), 1);
-  }, expected.length));
+    expectedMessages.splice(expectedMessages.indexOf(warning.message), 1);
+  }, expectedMessages.length);
+}
+
+function expectWarningByName(name, expected) {
+  if (typeof expected === 'string') {
+    expected = [expected];
+  }
+  process.on('warning', expectWarning(name, expected));
+}
+
+function expectWarningByMap(warningMap) {
+  var catchWarning = {};
+  forEach(objectKeys(warningMap), function (name) {
+    var expected = warningMap[name];
+    if (typeof expected === 'string') {
+      expected = [expected];
+    }
+    catchWarning[name] = expectWarning(name, expected);
+  });
+  process.on('warning', function (warning) {
+    return catchWarning[warning.name](warning);
+  });
+}
+
+// accepts a warning name and description or array of descriptions or a map
+// of warning names to description(s)
+// ensures a warning is generated for each name/description pair
+exports.expectWarning = function (nameOrMap, expected) {
+  if (typeof nameOrMap === 'string') {
+    expectWarningByName(nameOrMap, expected);
+  } else {
+    expectWarningByMap(nameOrMap);
+  }
 };
 
 /*<replacement>*/if (!process.browser) {

test/parallel/test-stream-duplex.js

@@ -3,7 +3,7 @@ var bufferShim = require('safe-buffer').Buffer;
 /*</replacement>*/
 require('../common');
 var assert = require('assert/');
-var Duplex = require('../../').Transform;
+var Duplex = require('../../').Duplex;
 
 var stream = new Duplex({ objectMode: true });
 

test/parallel/test-stream-unpipe-event.js

@@ -0,0 +1,93 @@
+/*<replacement>*/
+var bufferShim = require('safe-buffer').Buffer;
+/*</replacement>*/
+var common = require('../common');
+var assert = require('assert/');
+
+var _require = require('../../'),
+    Writable = _require.Writable,
+    Readable = _require.Readable;
+
+class NullWriteable extends Writable {
+  _write(chunk, encoding, callback) {
+    return callback();
+  }
+}
+class QuickEndReadable extends Readable {
+  _read() {
+    this.push(null);
+  }
+}
+class NeverEndReadable extends Readable {
+  _read() {}
+}
+
+function noop() {}
+
+{
+  var dest = new NullWriteable();
+  var src = new QuickEndReadable();
+  dest.on('pipe', common.mustCall(noop));
+  dest.on('unpipe', common.mustCall(noop));
+  src.pipe(dest);
+  setImmediate(function () {
+    assert.strictEqual(src._readableState.pipesCount, 0);
+  });
+}
+
+{
+  var _dest = new NullWriteable();
+  var _src = new NeverEndReadable();
+  _dest.on('pipe', common.mustCall(noop));
+  _dest.on('unpipe', common.mustNotCall('unpipe should not have been emitted'));
+  _src.pipe(_dest);
+  setImmediate(function () {
+    assert.strictEqual(_src._readableState.pipesCount, 1);
+  });
+}
+
+{
+  var _dest2 = new NullWriteable();
+  var _src2 = new NeverEndReadable();
+  _dest2.on('pipe', common.mustCall(noop));
+  _dest2.on('unpipe', common.mustCall(noop));
+  _src2.pipe(_dest2);
+  _src2.unpipe(_dest2);
+  setImmediate(function () {
+    assert.strictEqual(_src2._readableState.pipesCount, 0);
+  });
+}
+
+{
+  var _dest3 = new NullWriteable();
+  var _src3 = new QuickEndReadable();
+  _dest3.on('pipe', common.mustCall(noop));
+  _dest3.on('unpipe', common.mustCall(noop));
+  _src3.pipe(_dest3, { end: false });
+  setImmediate(function () {
+    assert.strictEqual(_src3._readableState.pipesCount, 0);
+  });
+}
+
+{
+  var _dest4 = new NullWriteable();
+  var _src4 = new NeverEndReadable();
+  _dest4.on('pipe', common.mustCall(noop));
+  _dest4.on('unpipe', common.mustNotCall('unpipe should not have been emitted'));
+  _src4.pipe(_dest4, { end: false });
+  setImmediate(function () {
+    assert.strictEqual(_src4._readableState.pipesCount, 1);
+  });
+}
+
+{
+  var _dest5 = new NullWriteable();
+  var _src5 = new NeverEndReadable();
+  _dest5.on('pipe', common.mustCall(noop));
+  _dest5.on('unpipe', common.mustCall(noop));
+  _src5.pipe(_dest5, { end: false });
+  _src5.unpipe(_dest5);
+  setImmediate(function () {
+    assert.strictEqual(_src5._readableState.pipesCount, 0);
+  });
+}