stream: sync writes don't need drain

Author: ronag

lib/_stream_writable.js

@@ -378,11 +378,6 @@ function writeOrBuffer(stream, state, isBuf, chunk, encoding, cb) {
 
   state.length += len;
 
-  const ret = state.length < state.highWaterMark;
-  // We must ensure that previous needDrain will not be reset to false.
-  if (!ret)
-    state.needDrain = true;
-
   if (state.writing || state.corked) {
     var last = state.lastBufferedRequest;
     state.lastBufferedRequest = {
@@ -402,6 +397,11 @@ function writeOrBuffer(stream, state, isBuf, chunk, encoding, cb) {
     doWrite(stream, state, false, len, chunk, encoding, cb);
   }
 
+  const ret = state.length < state.highWaterMark;
+  // We must ensure that previous needDrain will not be reset to false.
+  if (!ret)
+    state.needDrain = true;
+
   return ret;
 }
 

test/parallel/test-stream-big-packet.js

@@ -32,12 +32,19 @@ class TestStream extends stream.Transform {
       // Char 'a' only exists in the last write
       passed = chunk.toString().includes('a');
     }
-    done();
+    process.nextTick(done);
   }
 }
 
-const s1 = new stream.PassThrough();
-const s2 = new stream.PassThrough();
+class PassStream extends stream.Transform {
+  _transform(chunk, encoding, done) {
+    this.push(chunk);
+    process.nextTick(done);
+  }
+}
+
+const s1 = new PassStream();
+const s2 = new PassStream();
 const s3 = new TestStream();
 s1.pipe(s3);
 // Don't let s2 auto close which may close s3

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

@@ -19,7 +19,7 @@ const writable = new stream.Writable({
       });
     }
 
-    cb();
+    process.nextTick(cb);
   }, 3)
 });
 

test/parallel/test-stream-pipe-await-drain.js

@@ -4,11 +4,16 @@ const stream = require('stream');
 const assert = require('assert');
 
 // This is very similar to test-stream-pipe-cleanup-pause.js.
+class TestStream extends stream.Writable {
+  _write(chunk, encoding, done) {
+    process.nextTick(done);
+  }
+}
 
 const reader = new stream.Readable();
-const writer1 = new stream.Writable();
-const writer2 = new stream.Writable();
-const writer3 = new stream.Writable();
+const writer1 = new TestStream();
+const writer2 = new TestStream();
+const writer3 = new TestStream();
 
 // 560000 is chosen here because it is larger than the (default) highWaterMark
 // and will cause `.write()` to return false
@@ -19,7 +24,7 @@ reader._read = () => {};
 
 writer1._write = common.mustCall(function(chunk, encoding, cb) {
   this.emit('chunk-received');
-  cb();
+  process.nextTick(cb);
 }, 1);
 
 writer1.once('chunk-received', () => {

test/parallel/test-stream-writable-needdrain-state.js

@@ -4,20 +4,49 @@ const common = require('../common');
 const stream = require('stream');
 const assert = require('assert');
 
-const transform = new stream.Transform({
-  transform: _transform,
-  highWaterMark: 1
-});
-
-function _transform(chunk, encoding, cb) {
-  assert.strictEqual(transform._writableState.needDrain, true);
-  cb();
-}
+{
+  const transform = new stream.Transform({
+    transform: _transform,
+    highWaterMark: 1
+  });
+
+  function _transform(chunk, encoding, cb) {
+    assert.strictEqual(transform._writableState.needDrain, false);
+    cb();
+  }
+
+  assert.strictEqual(transform._writableState.needDrain, false);
 
-assert.strictEqual(transform._writableState.needDrain, false);
+  transform.write('asdasd', common.mustCall(() => {
+    assert.strictEqual(transform._writableState.needDrain, false);
+  }));
 
-transform.write('asdasd', common.mustCall(() => {
   assert.strictEqual(transform._writableState.needDrain, false);
-}));
+}
+
+{
+  const w = new stream.Writable({
+    highWaterMark: 1
+  });
+
+  w._write = (chunk, encoding, cb) => {
+    cb();
+  };
+  w.on('drain', common.mustNotCall());
 
-assert.strictEqual(transform._writableState.needDrain, true);
+  assert.strictEqual(w.write('asd'), true);
+}
+
+
+{
+  const w = new stream.Writable({
+    highWaterMark: 1
+  });
+
+  w._write = (chunk, encoding, cb) => {
+    process.nextTick(cb);
+  };
+  w.on('drain', common.mustCall());
+
+  assert.strictEqual(w.write('asd'), false);
+}