stream: writableNeedDrain

Don't write to a stream which already has a full buffer.

Fixes: #35341

PR-URL: #35348
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Luigi Pinca <[email protected]>
Reviewed-By: Benjamin Gruenbaum <[email protected]>

doc/api/stream.md

@@ -580,6 +580,15 @@ This property contains the number of bytes (or objects) in the queue
 ready to be written. The value provides introspection data regarding
 the status of the `highWaterMark`.
 
+##### `writable.writableNeedDrain`
+<!-- YAML
+added: REPLACEME
+-->
+
+* {boolean}
+
+Is `true` if the stream's buffer has been full and stream will emit `'drain'`.
+
 ##### `writable.writableObjectMode`
 <!-- YAML
 added: v12.3.0

lib/_http_outgoing.js

@@ -660,6 +660,11 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'writableEnded', {
   get: function() { return this.finished; }
 });
 
+ObjectDefineProperty(OutgoingMessage.prototype, 'writableNeedDrain', {
+  get: function() {
+    return !this.destroyed && !this.finished && this[kNeedDrain];
+  }
+});
 
 const crlf_buf = Buffer.from('\r\n');
 OutgoingMessage.prototype.write = function write(chunk, encoding, callback) {

lib/internal/streams/duplex.js

@@ -87,6 +87,8 @@ ObjectDefineProperties(Duplex.prototype, {
     ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableCorked'),
   writableEnded:
     ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableEnded'),
+  writableNeedDrain:
+    ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableNeedDrain'),
 
   destroyed: {
     get() {

lib/internal/streams/pipeline.js

@@ -123,6 +123,10 @@ async function pump(iterable, writable, finish) {
   }
   let error;
   try {
+    if (writable.writableNeedDrain === true) {
+      await EE.once(writable, 'drain');
+    }
+
     for await (const chunk of iterable) {
       if (!writable.write(chunk)) {
         if (writable.destroyed) return;

lib/internal/streams/readable.js

@@ -783,7 +783,12 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   dest.emit('pipe', src);
 
   // Start the flow if it hasn't been started already.
-  if (!state.flowing) {
+
+  if (dest.writableNeedDrain === true) {
+    if (state.flowing) {
+      src.pause();
+    }
+  } else if (!state.flowing) {
     debug('pipe resume');
     src.resume();
   }

lib/internal/streams/writable.js

@@ -805,6 +805,14 @@ ObjectDefineProperties(Writable.prototype, {
     }
   },
 
+  writableNeedDrain: {
+    get() {
+      const wState = this._writableState;
+      if (!wState) return false;
+      return !wState.destroyed && !wState.ending && wState.needDrain;
+    }
+  },
+
   writableHighWaterMark: {
     get() {
       return this._writableState && this._writableState.highWaterMark;

test/parallel/test-stream-pipe-needDrain.js

@@ -0,0 +1,29 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const Readable = require('_stream_readable');
+const Writable = require('_stream_writable');
+
+// Pipe should not continue writing if writable needs drain.
+{
+  const w = new Writable({
+    write(buf, encoding, callback) {
+
+    }
+  });
+
+  while (w.write('asd'));
+
+  assert.strictEqual(w.writableNeedDrain, true);
+
+  const r = new Readable({
+    read() {
+      this.push('asd');
+    }
+  });
+
+  w.write = common.mustNotCall();
+
+  r.pipe(w);
+}