stream: use lazy registration for drain for fast destinations

PR-URL: https://github.com/nodejs/node/pull/29095
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

1 files changed, 13 insertions, 8 deletions

diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index 1830eea66d..01f95104b8 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -676,12 +676,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
     dest.end();
   }
 
-  // When the dest drains, it reduces the awaitDrain counter
-  // on the source.  This would be more elegant with a .once()
-  // handler in flow(), but adding and removing repeatedly is
-  // too slow.
-  const ondrain = pipeOnDrain(src);
-  dest.on('drain', ondrain);
+  let ondrain;
 
   var cleanedUp = false;
   function cleanup() {
@@ -689,7 +684,9 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
     // Cleanup event handlers once the pipe is broken
     dest.removeListener('close', onclose);
     dest.removeListener('finish', onfinish);
-    dest.removeListener('drain', ondrain);
+    if (ondrain) {
+      dest.removeListener('drain', ondrain);
+    }
     dest.removeListener('error', onerror);
     dest.removeListener('unpipe', onunpipe);
     src.removeListener('end', onend);
@@ -703,7 +700,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
     // flowing again.
     // So, if this is awaiting a drain, then we just call it now.
     // If we don't know, then assume that we are waiting for one.
-    if (state.awaitDrain &&
+    if (ondrain && state.awaitDrain &&
         (!dest._writableState || dest._writableState.needDrain))
       ondrain();
   }
@@ -722,6 +719,14 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
         debug('false write response, pause', state.awaitDrain);
         state.awaitDrain++;
       }
+      if (!ondrain) {
+        // When the dest drains, it reduces the awaitDrain counter
+        // on the source.  This would be more elegant with a .once()
+        // handler in flow(), but adding and removing repeatedly is
+        // too slow.
+        ondrain = pipeOnDrain(src);
+        dest.on('drain', ondrain);
+      }
       src.pause();
     }
   }