Update to node master

lib/_stream_readable.js

@@ -25,6 +25,7 @@ Readable.ReadableState = ReadableState;
 var Stream = require('stream');
 var util = require('util');
 var StringDecoder;
+var Buffer = require('buffer').Buffer;
 
 util.inherits(Readable, Stream);
 
@@ -137,8 +138,13 @@ function howMuchToRead(n, state) {
   if (state.objectMode)
     return n === 0 ? 0 : 1;
 
-  if (isNaN(n) || n === null)
-    return state.length;
+  if (isNaN(n) || n === null) {
+    // only flow one buffer at a time
+    if (state.flowing && state.buffer.length)
+      return state.buffer[0].length;
+    else
+      return state.length;
+  }
 
   if (n <= 0)
     return 0;
@@ -163,6 +169,16 @@ Readable.prototype.read = function(n) {
   if (typeof n !== 'number' || n > 0)
     state.emittedReadable = false;
 
+  // if we're doing read(0) to trigger a readable event, but we
+  // already have a bunch of data in the buffer, then just trigger
+  // the 'readable' event and move on.
+  if (n === 0 &&
+      state.needReadable &&
+      state.length >= state.highWaterMark) {
+    emitReadable(this);
+    return null;
+  }
+
   n = howMuchToRead(n, state);
 
   // if we've ended, and we're now clear, then finish it up.
@@ -229,7 +245,7 @@ Readable.prototype.read = function(n) {
   else
     ret = null;
 
-  if (ret === null || (!state.objectMode && ret.length === 0)) {
+  if (ret === null) {
     state.needReadable = true;
     n = 0;
   }
@@ -275,7 +291,7 @@ function onread(stream, er, chunk) {
     // eof
     state.ended = true;
     if (state.decoder) {
-      chunk = state.decoder.end();
+      chunk = endStringDecoder(state.decoder);
       if (chunk && chunk.length) {
         state.buffer.push(chunk);
         state.length += state.objectMode ? 1 : chunk.length;
@@ -670,7 +686,7 @@ Readable.prototype.wrap = function(stream) {
   stream.on('end', function() {
     state.ended = true;
     if (state.decoder) {
-      var chunk = state.decoder.end();
+      chunk = endStringDecoder(state.decoder);
       if (chunk && chunk.length)
         self.push(chunk);
     }
@@ -806,3 +822,20 @@ function endReadable(stream) {
     stream.emit('end');
   });
 }
+
+function endStringDecoder(decoder) {
+  if (decoder.end) {
+    return decoder.end()
+  }
+
+  var res = '';
+
+  if (decoder.charReceived) {
+    var cr = decoder.charReceived;
+    var buf = decoder.charBuffer;
+    var enc = decoder.encoding;
+    res += buf.slice(0, cr).toString(enc);
+  }
+
+  return res;
+}

lib/_stream_writable.js

@@ -30,6 +30,7 @@ var util = require('util');
 var assert = require('assert');
 var Stream = require('stream');
 var Duplex = require('./_stream_duplex');
+var Buffer = require('buffer').Buffer;
 
 util.inherits(Writable, Stream);
 

test/simple/test-stream2-basic.js

@@ -21,7 +21,7 @@
 
 
 var common = require('../common.js');
-var R = require('../../lib/_stream_readable.js');
+var R = require('../../readable');
 var assert = require('assert');
 
 var util = require('util');

test/simple/test-stream2-fs.js → test/simple/test-stream2-compatibility.js

@@ -24,53 +24,27 @@ var common = require('../common.js');
 var R = require('../../readable');
 var assert = require('assert');
 
-var fs = require('../../fs');
-var FSReadable = fs.ReadStream;
-
-var path = require('path');
-var file = path.resolve(common.fixturesDir, 'x1024.txt');
-
-var size = fs.statSync(file).size;
-
-// expect to see chunks no more than 10 bytes each.
-var expectLengths = [];
-for (var i = size; i > 0; i -= 10) {
-  expectLengths.push(Math.min(i, 10));
-}
-
 var util = require('util');
-var Stream = require('stream');
+var EE = require('events').EventEmitter;
 
-util.inherits(TestWriter, Stream);
+var ondataCalled = 0;
 
-function TestWriter() {
-  Stream.apply(this);
-  this.buffer = [];
-  this.length = 0;
-}
-
-TestWriter.prototype.write = function(c) {
-  this.buffer.push(c.toString());
-  this.length += c.length;
-  return true;
-};
+function TestReader() {
+  R.apply(this);
+  this._buffer = new Buffer(100);
+  this._buffer.fill('x');
 
-TestWriter.prototype.end = function(c) {
-  if (c) this.buffer.push(c.toString());
-  this.emit('results', this.buffer);
+  this.on('data', function() {
+    ondataCalled++;
+  });
 }
 
-var r = new FSReadable(file, { bufferSize: 10 });
-var w = new TestWriter();
+util.inherits(TestReader, R);
 
-w.on('results', function(res) {
-  console.error(res, w.length);
-  assert.equal(w.length, size);
-  var l = 0;
-  assert.deepEqual(res.map(function (c) {
-    return c.length;
-  }), expectLengths);
-  console.log('ok');
-});
+TestReader.prototype._read = function(n, cb) {
+  cb(null, this._buffer);
+  this._buffer = new Buffer(0);
+};
 
-r.pipe(w, { chunkSize: 10 });
+var reader = new TestReader();
+assert.equal(ondataCalled, 1);

test/simple/test-stream2-finish-pipe.js

@@ -0,0 +1,42 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+var common = require('../common.js');
+var Readable = require('../../readable');
+var Writable = require('../../writable')
+var Buffer = require('buffer').Buffer;
+
+var R = new Readable();
+R._read = function(size, cb) {
+  cb(null, new Buffer(size));
+};
+
+var W = new Writable();
+W._write = function(data, cb) {
+  cb(null);
+};
+
+R.pipe(W);
+
+// This might sound unrealistic, but it happens in net.js. When
+// `socket.allowHalfOpen === false`, EOF will cause `.destroySoon()` call which
+// ends the writable side of net.Socket.
+W.end();

test/simple/test-stream2-large-read-stall.js

@@ -0,0 +1,82 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+var common = require('../common.js');
+var assert = require('assert');
+
+// If everything aligns so that you do a read(n) of exactly the
+// remaining buffer, then make sure that 'end' still emits.
+
+var READSIZE = 100;
+var PUSHSIZE = 20;
+var PUSHCOUNT = 1000;
+var HWM = 50;
+
+var Readable = require('../../readable');
+var r = new Readable({
+  highWaterMark: HWM
+});
+var rs = r._readableState;
+
+r._read = push;
+
+r.on('readable', function() {
+  console.error('>> readable');
+  do {
+    console.error('  > read(%d)', READSIZE);
+    var ret = r.read(READSIZE);
+    console.error('  < %j (%d remain)', ret && ret.length, rs.length);
+  } while (ret && ret.length === READSIZE);
+
+  console.error('<< after read()',
+                ret && ret.length,
+                rs.needReadable,
+                rs.length);
+});
+
+var endEmitted = false;
+r.on('end', function() {
+  endEmitted = true;
+  console.error('end');
+});
+
+var pushes = 0;
+function push() {
+  if (pushes > PUSHCOUNT)
+    return;
+
+  if (pushes++ === PUSHCOUNT) {
+    console.error('   push(EOF)');
+    return r.push(null);
+  }
+
+  console.error('   push #%d', pushes);
+  if (r.push(new Buffer(PUSHSIZE)))
+    setTimeout(push);
+}
+
+// start the flow
+var ret = r.read(0);
+
+process.on('exit', function() {
+  assert.equal(pushes, PUSHCOUNT + 1);
+  assert(endEmitted);
+});