stream: always reset awaitDrain when emitting data

lib/_stream_readable.js

@@ -258,6 +258,7 @@ function readableAddChunk(stream, chunk, encoding, addToFront, skipChunkCheck) {
 
 function addChunk(stream, state, chunk, addToFront) {
   if (state.flowing && state.length === 0 && !state.sync) {
+    state.awaitDrain = 0;
     stream.emit('data', chunk);
   } else {
     // update the buffer info.
@@ -456,6 +457,7 @@ Readable.prototype.read = function(n) {
     n = 0;
   } else {
     state.length -= n;
+    state.awaitDrain = 0;
   }
 
   if (state.length === 0) {
@@ -637,18 +639,12 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
       ondrain();
   }
 
-  // If the user pushes more data while we're writing to dest then we'll end up
-  // in ondata again. However, we only want to increase awaitDrain once because
-  // dest will only emit one 'drain' event for the multiple writes.
-  // => Introduce a guard on increasing awaitDrain.
-  var increasedAwaitDrain = false;
   src.on('data', ondata);
   function ondata(chunk) {
     debug('ondata');
-    increasedAwaitDrain = false;
     var ret = dest.write(chunk);
     debug('dest.write', ret);
-    if (false === ret && !increasedAwaitDrain) {
+    if (ret === false) {
       // If the user unpiped during `dest.write()`, it is possible
       // to get stuck in a permanently paused state if that write
       // also returned false.
@@ -658,7 +654,6 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
           !cleanedUp) {
         debug('false write response, pause', state.awaitDrain);
         state.awaitDrain++;
-        increasedAwaitDrain = true;
       }
       src.pause();
     }
@@ -834,7 +829,6 @@ function resume_(stream, state) {
   }
 
   state.resumeScheduled = false;
-  state.awaitDrain = 0;
   stream.emit('resume');
   flow(stream);
   if (state.flowing && !state.reading)

test/parallel/test-stream-pipe-flow-after-unpipe.js

@@ -0,0 +1,29 @@
+'use strict';
+const common = require('../common');
+const { Readable, Writable } = require('stream');
+
+// Tests that calling .unpipe() un-blocks a stream that is paused because
+// it is waiting on the writable side to finish a write().
+
+const rs = new Readable({
+  highWaterMark: 1,
+  // That this gets called at least 20 times is the real test here.
+  read: common.mustCallAtLeast(() => rs.push('foo'), 20)
+});
+
+const ws = new Writable({
+  highWaterMark: 1,
+  write: common.mustCall(() => {
+    // Ignore the callback, this write() simply never finishes.
+    setImmediate(() => rs.unpipe(ws));
+  })
+});
+
+let chunks = 0;
+rs.on('data', common.mustCallAtLeast(() => {
+  chunks++;
+  if (chunks >= 20)
+    rs.pause();  // Finish this test.
+}));
+
+rs.pipe(ws);

test/parallel/test-stream-pipe-manual-resume.js

@@ -0,0 +1,35 @@
+'use strict';
+const common = require('../common');
+const stream = require('stream');
+
+function test(throwCodeInbetween) {
+  // Check that a pipe does not stall if .read() is called unexpectedly
+  // (i.e. the stream is not resumed by the pipe).
+
+  const n = 1000;
+  let counter = n;
+  const rs = stream.Readable({
+    objectMode: true,
+    read: common.mustCallAtLeast(() => {
+      if (--counter >= 0)
+        rs.push({ counter });
+      else
+        rs.push(null);
+    }, n)
+  });
+
+  const ws = stream.Writable({
+    objectMode: true,
+    write: common.mustCall((data, enc, cb) => {
+      setImmediate(cb);
+    }, n)
+  });
+
+  setImmediate(() => throwCodeInbetween(rs, ws));
+
+  rs.pipe(ws);
+}
+
+test((rs) => rs.read());
+test((rs) => rs.resume());
+test(() => 0);