streams: Pre-emptively buffer readables up to the highWaterMark

Also, this adds a test that guarantees that the ordering of several
push() calls in a row is always preserved in synchronous readable streams

lib/_stream_readable.js

@@ -329,6 +329,12 @@ function onread(stream, er, chunk) {
 
   if (state.needReadable)
     emitReadable(stream);
+  else if (state.sync)
+    process.nextTick(function() {
+      maybeReadMore(stream, state);
+    });
+  else
+    maybeReadMore(stream, state);
 }
 
 // Don't emit readable right away in sync mode, because this can trigger
@@ -352,7 +358,10 @@ function emitReadable(stream) {
 function emitReadable_(stream) {
   var state = stream._readableState;
   stream.emit('readable');
+  maybeReadMore(stream, state);
+}
 
+function maybeReadMore(stream, state) {
   // at this point, the user has presumably seen the 'readable' event,
   // and called read() to consume some data.  that may have triggered
   // in turn another _read(n,cb) call, in which case reading = true if

test/simple/test-stream-push-order.js

@@ -0,0 +1,51 @@
+// 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('stream').Readable;
+var assert = require('assert');
+
+var s = new Readable({
+  highWaterMark: 20,
+  encoding: 'ascii'
+});
+
+var list = ['1', '2', '3', '4', '5', '6'];
+
+s._read = function (n, cb) {
+  var one = list.shift();
+  if (!one)
+    return cb(null, null);
+
+  var two = list.shift();
+  s.push(one);
+  cb(null, two);
+};
+
+var v = s.read(0);
+
+// ACTUALLY [1, 3, 5, 6, 4, 2]
+
+process.on("exit", function () {
+  assert.deepEqual(s._readableState.buffer,
+                   ['1', '2', '3', '4', '5', '6']);
+  console.log("ok");
+});

test/simple/test-stream2-basic.js

@@ -334,13 +334,13 @@ test('back pressure respected', function (t) {
   function noop() {}
 
   var r = new R();
+  r._read = noop;
   var counter = 0;
   r.push(["one"]);
   r.push(["two"]);
   r.push(["three"]);
   r.push(["four"]);
   r.push(null);
-  r._read = noop;
 
   var w1 = new R();
   w1.write = function (chunk) {

test/simple/test-stream2-objects.js

@@ -75,11 +75,11 @@ function toArray(callback) {
 
 function fromArray(list) {
   var r = new Readable();
+  r._read = noop;
   list.forEach(function(chunk) {
     r.push(chunk);
   });
   r.push(null);
-  r._read = noop;
 
   return r;
 }