Updated from Node 8.1.3.

README.md

@@ -1,6 +1,6 @@
 # readable-stream
 
-***Node-core v8.1.2 streams for userland*** [![Build Status](https://travis-ci.org/nodejs/readable-stream.svg?branch=master)](https://travis-ci.org/nodejs/readable-stream)
+***Node-core v8.1.3 streams for userland*** [![Build Status](https://travis-ci.org/nodejs/readable-stream.svg?branch=master)](https://travis-ci.org/nodejs/readable-stream)
 
 
 [![NPM](https://nodei.co/npm/readable-stream.png?downloads=true&downloadRank=true)](https://nodei.co/npm/readable-stream/)
@@ -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/v8.1.2/docs/api/stream.html).
+Full documentation may be found on the [Node.js website](https://nodejs.org/dist/v8.1.3/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

@@ -254,7 +254,7 @@ function readableAddChunk(stream, chunk, encoding, addToFront, skipChunkCheck) {
     if (er) {
       stream.emit('error', er);
     } else if (state.objectMode || chunk && chunk.length > 0) {
-      if (typeof chunk !== 'string' && Object.getPrototypeOf(chunk) !== Buffer.prototype && !state.objectMode) {
+      if (typeof chunk !== 'string' && !state.objectMode && Object.getPrototypeOf(chunk) !== Buffer.prototype) {
         chunk = _uint8ArrayToBuffer(chunk);
       }
 

lib/_stream_writable.js

@@ -409,15 +409,26 @@ function doWrite(stream, state, writev, len, chunk, encoding, cb) {
 
 function onwriteError(stream, state, sync, er, cb) {
   --state.pendingcb;
-  if (sync) processNextTick(afterError, stream, state, cb, er);else afterError(stream, state, cb, er);
 
-  stream._writableState.errorEmitted = true;
-  stream.emit('error', er);
-}
-
-function afterError(stream, state, cb, err) {
-  cb(err);
-  finishMaybe(stream, state);
+  if (sync) {
+    // defer the callback if we are being called synchronously
+    // to avoid piling up things on the stack
+    processNextTick(cb, er);
+    // this can emit finish, and it will always happen
+    // after error
+    processNextTick(finishMaybe, stream, state);
+    stream._writableState.errorEmitted = true;
+    stream.emit('error', er);
+  } else {
+    // the caller expect this to happen before if
+    // it is async
+    cb(er);
+    stream._writableState.errorEmitted = true;
+    stream.emit('error', er);
+    // this can emit finish, but finish must
+    // always follow error
+    finishMaybe(stream, state);
+  }
 }
 
 function onwriteStateUpdate(state) {

package.json

@@ -8,8 +8,8 @@
     "inherits": "~2.0.3",
     "isarray": "~1.0.0",
     "process-nextick-args": "~1.0.6",
-    "safe-buffer": "~5.1.0",
-    "string_decoder": "~1.0.0",
+    "safe-buffer": "~5.1.1",
+    "string_decoder": "~1.0.3",
     "util-deprecate": "~1.0.1"
   },
   "devDependencies": {

test/common.js

@@ -292,11 +292,11 @@ if (exports.isWindows) {
 }
 
 var ifaces = os.networkInterfaces();
+var re = /lo/;
 exports.hasIPv6 = objectKeys(ifaces).some(function (name) {
-  return (/lo/.test(name) && ifaces[name].some(function (info) {
-      return info.family === 'IPv6';
-    })
-  );
+  return re.test(name) && ifaces[name].some(function (info) {
+    return info.family === 'IPv6';
+  });
 });
 
 /*
@@ -454,7 +454,7 @@ function leakedGlobals() {
     if (!knownGlobals.includes(global[val])) leaked.push(val);
   }if (global.__coverage__) {
     return leaked.filter(function (varname) {
-      return !/^(cov_|__cov)/.test(varname);
+      return !/^(?:cov_|__cov)/.test(varname);
     });
   } else {
     return leaked;
@@ -703,6 +703,14 @@ exports.expectWarning = function (nameOrMap, expected) {
   });
 } /*</replacement>*/
 
+/*<replacement>*/if (!process.browser) {
+  Object.defineProperty(exports, 'hasSmallICU', {
+    get: function () {
+      return process.binding('config').hasSmallICU;
+    }
+  });
+} /*</replacement>*/
+
 // Useful for testing expected internal/error objects
 exports.expectsError = function expectsError(_ref) {
   var code = _ref.code,

test/parallel/test-stream-objectmode-undefined.js

@@ -0,0 +1,50 @@
+/*<replacement>*/
+var bufferShim = require('safe-buffer').Buffer;
+/*</replacement>*/
+var common = require('../common');
+var assert = require('assert/');
+
+var _require = require('../../'),
+    Readable = _require.Readable,
+    Writable = _require.Writable,
+    Transform = _require.Transform;
+
+{
+  var stream = new Readable({
+    objectMode: true,
+    read: common.mustCall(function () {
+      stream.push(undefined);
+      stream.push(null);
+    })
+  });
+
+  stream.on('data', common.mustCall(function (chunk) {
+    assert.strictEqual(chunk, undefined);
+  }));
+}
+
+{
+  var _stream = new Writable({
+    objectMode: true,
+    write: common.mustCall(function (chunk) {
+      assert.strictEqual(chunk, undefined);
+    })
+  });
+
+  _stream.write(undefined);
+}
+
+{
+  var _stream2 = new Transform({
+    objectMode: true,
+    transform: common.mustCall(function (chunk) {
+      _stream2.push(chunk);
+    })
+  });
+
+  _stream2.on('data', common.mustCall(function (chunk) {
+    assert.strictEqual(chunk, undefined);
+  }));
+
+  _stream2.write(undefined);
+}

test/parallel/test-stream-pipe-await-drain-push-while-write.js

@@ -19,7 +19,7 @@ var writable = new stream.Writable({
     if (chunk.length === 32 * 1024) {
       // first chunk
       var beforePush = readable._readableState.awaitDrain;
-      readable.push(new Buffer(34 * 1024)); // above hwm
+      readable.push(bufferShim.alloc(34 * 1024)); // above hwm
       // We should check if awaitDrain counter is increased.
       var afterPush = readable._readableState.awaitDrain;
       assert.strictEqual(afterPush - beforePush, 1, 'Counter is not increased for awaitDrain');
@@ -31,7 +31,7 @@ var writable = new stream.Writable({
 });
 
 // A readable stream which produces two buffers.
-var bufs = [new Buffer(32 * 1024), new Buffer(33 * 1024)]; // above hwm
+var bufs = [bufferShim.alloc(32 * 1024), bufferShim.alloc(33 * 1024)]; // above hwm
 var readable = new stream.Readable({
   read: function () {
     while (bufs.length > 0) {

test/parallel/test-stream-readable-invalid-chunk.js

@@ -10,12 +10,13 @@ var readable = new stream.Readable({
   read: common.noop
 });
 
+var errMessage = /Invalid non-string\/buffer chunk/;
 assert.throws(function () {
   return readable.push([]);
-}, /Invalid non-string\/buffer chunk/);
+}, errMessage);
 assert.throws(function () {
   return readable.push({});
-}, /Invalid non-string\/buffer chunk/);
+}, errMessage);
 assert.throws(function () {
   return readable.push(0);
-}, /Invalid non-string\/buffer chunk/);
+}, errMessage);

test/parallel/test-stream-writable-write-writev-finish.js

@@ -0,0 +1,158 @@
+/*<replacement>*/
+var bufferShim = require('safe-buffer').Buffer;
+/*</replacement>*/
+
+var common = require('../common');
+var assert = require('assert/');
+var stream = require('../../');
+
+// ensure consistency between the finish event when using cork()
+// and writev and when not using them
+
+{
+  var writable = new stream.Writable();
+
+  writable._write = function (chunks, encoding, cb) {
+    cb(new Error('write test error'));
+  };
+
+  var firstError = false;
+  writable.on('finish', common.mustCall(function () {
+    assert.strictEqual(firstError, true);
+  }));
+
+  writable.on('prefinish', common.mustCall());
+
+  writable.on('error', common.mustCall(function (er) {
+    assert.strictEqual(er.message, 'write test error');
+    firstError = true;
+  }));
+
+  writable.end('test');
+}
+
+{
+  var _writable = new stream.Writable();
+
+  _writable._write = function (chunks, encoding, cb) {
+    setImmediate(cb, new Error('write test error'));
+  };
+
+  var _firstError = false;
+  _writable.on('finish', common.mustCall(function () {
+    assert.strictEqual(_firstError, true);
+  }));
+
+  _writable.on('prefinish', common.mustCall());
+
+  _writable.on('error', common.mustCall(function (er) {
+    assert.strictEqual(er.message, 'write test error');
+    _firstError = true;
+  }));
+
+  _writable.end('test');
+}
+
+{
+  var _writable2 = new stream.Writable();
+
+  _writable2._write = function (chunks, encoding, cb) {
+    cb(new Error('write test error'));
+  };
+
+  _writable2._writev = function (chunks, cb) {
+    cb(new Error('writev test error'));
+  };
+
+  var _firstError2 = false;
+  _writable2.on('finish', common.mustCall(function () {
+    assert.strictEqual(_firstError2, true);
+  }));
+
+  _writable2.on('prefinish', common.mustCall());
+
+  _writable2.on('error', common.mustCall(function (er) {
+    assert.strictEqual(er.message, 'writev test error');
+    _firstError2 = true;
+  }));
+
+  _writable2.cork();
+  _writable2.write('test');
+
+  setImmediate(function () {
+    _writable2.end('test');
+  });
+}
+
+{
+  var _writable3 = new stream.Writable();
+
+  _writable3._write = function (chunks, encoding, cb) {
+    setImmediate(cb, new Error('write test error'));
+  };
+
+  _writable3._writev = function (chunks, cb) {
+    setImmediate(cb, new Error('writev test error'));
+  };
+
+  var _firstError3 = false;
+  _writable3.on('finish', common.mustCall(function () {
+    assert.strictEqual(_firstError3, true);
+  }));
+
+  _writable3.on('prefinish', common.mustCall());
+
+  _writable3.on('error', common.mustCall(function (er) {
+    assert.strictEqual(er.message, 'writev test error');
+    _firstError3 = true;
+  }));
+
+  _writable3.cork();
+  _writable3.write('test');
+
+  setImmediate(function () {
+    _writable3.end('test');
+  });
+}
+
+// Regression test for
+// https://github.com/nodejs/node/issues/13812
+
+{
+  var rs = new stream.Readable();
+  rs.push('ok');
+  rs.push(null);
+  rs._read = function () {};
+
+  var ws = new stream.Writable();
+  var _firstError4 = false;
+
+  ws.on('finish', common.mustCall(function () {
+    assert.strictEqual(_firstError4, true);
+  }));
+  ws.on('error', common.mustCall(function () {
+    _firstError4 = true;
+  }));
+
+  ws._write = function (chunk, encoding, done) {
+    setImmediate(done, new Error());
+  };
+  rs.pipe(ws);
+}
+
+{
+  var _rs = new stream.Readable();
+  _rs.push('ok');
+  _rs.push(null);
+  _rs._read = function () {};
+
+  var _ws = new stream.Writable();
+
+  _ws.on('finish', common.mustNotCall());
+  _ws.on('error', common.mustCall());
+
+  _ws._write = function (chunk, encoding, done) {
+    done(new Error());
+  };
+  _rs.pipe(_ws);
+}