stream: remove thenable support

Remove support for returning thenables in stream
implementation methods. This is causing more confusion
and issues than it's worth.

Refs: https://github.com/nodejs/node/issues/39535

PR-URL: https://github.com/nodejs/node/pull/40773
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>

4 files changed, 6 insertions, 127 deletions

diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
index efa09e05eaf..10f5471e21d 100644
--- a/lib/internal/streams/destroy.js
+++ b/lib/internal/streams/destroy.js
@@ -106,20 +106,7 @@ function _destroy(self, err, cb) {
     }
   }
   try {
-    const result = self._destroy(err || null, onDestroy);
-    if (result != null) {
-      const then = result.then;
-      if (typeof then === 'function') {
-        then.call(
-          result,
-          function() {
-            process.nextTick(onDestroy, null);
-          },
-          function(err) {
-            process.nextTick(onDestroy, err);
-          });
-      }
-    }
+    self._destroy(err || null, onDestroy);
   } catch (err) {
     onDestroy(err);
   }
@@ -285,24 +272,7 @@ function constructNT(stream) {
   }
 
   try {
-    const result = stream._construct(onConstruct);
-    if (result != null) {
-      const then = result.then;
-      if (typeof then === 'function') {
-        then.call(
-          result,
-          function() {
-            if (!called) {
-              process.nextTick(onConstruct, null);
-            }
-          },
-          function(err) {
-            if (!called) {
-              process.nextTick(onConstruct, err);
-            }
-          });
-      }
-    }
+    stream._construct(onConstruct);
   } catch (err) {
     onConstruct(err);
   }
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index 4b7613fd13d..8081e8fde13 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -493,18 +493,7 @@ Readable.prototype.read = function(n) {
 
     // Call internal read method
     try {
-      const result = this._read(state.highWaterMark);
-      if (result != null) {
-        const then = result.then;
-        if (typeof then === 'function') {
-          then.call(
-            result,
-            nop,
-            function(err) {
-              errorOrDestroy(this, err);
-            });
-        }
-      }
+      this._read(state.highWaterMark);
     } catch (err) {
       errorOrDestroy(this, err);
     }
diff --git a/lib/internal/streams/transform.js b/lib/internal/streams/transform.js
index cbd23185fad..fdac76e4062 100644
--- a/lib/internal/streams/transform.js
+++ b/lib/internal/streams/transform.js
@@ -107,10 +107,8 @@ function Transform(options) {
 }
 
 function final(cb) {
-  let called = false;
   if (typeof this._flush === 'function' && !this.destroyed) {
-    const result = this._flush((er, data) => {
-      called = true;
+    this._flush((er, data) => {
       if (er) {
         if (cb) {
           cb(er);
@@ -128,33 +126,6 @@ function final(cb) {
         cb();
       }
     });
-    if (result !== undefined && result !== null) {
-      try {
-        const then = result.then;
-        if (typeof then === 'function') {
-          then.call(
-            result,
-            (data) => {
-              if (called)
-                return;
-              if (data != null)
-                this.push(data);
-              this.push(null);
-              if (cb)
-                process.nextTick(cb);
-            },
-            (err) => {
-              if (cb) {
-                process.nextTick(cb, err);
-              } else {
-                process.nextTick(() => this.destroy(err));
-              }
-            });
-        }
-      } catch (err) {
-        process.nextTick(() => this.destroy(err));
-      }
-    }
   } else {
     this.push(null);
     if (cb) {
@@ -180,9 +151,7 @@ Transform.prototype._write = function(chunk, encoding, callback) {
   const wState = this._writableState;
   const length = rState.length;
 
-  let called = false;
-  const result = this._transform(chunk, encoding, (err, val) => {
-    called = true;
+  this._transform(chunk, encoding, (err, val) => {
     if (err) {
       callback(err);
       return;
@@ -204,38 +173,6 @@ Transform.prototype._write = function(chunk, encoding, callback) {
       this[kCallback] = callback;
     }
   });
-  if (result !== undefined && result != null) {
-    try {
-      const then = result.then;
-      if (typeof then === 'function') {
-        then.call(
-          result,
-          (val) => {
-            if (called)
-              return;
-
-            if (val != null) {
-              this.push(val);
-            }
-
-            if (
-              wState.ended ||
-              length === rState.length ||
-              rState.length < rState.highWaterMark ||
-              rState.length === 0) {
-              process.nextTick(callback);
-            } else {
-              this[kCallback] = callback;
-            }
-          },
-          (err) => {
-            process.nextTick(callback, err);
-          });
-      }
-    } catch (err) {
-      process.nextTick(callback, err);
-    }
-  }
 };
 
 Transform.prototype._read = function() {
diff --git a/lib/internal/streams/writable.js b/lib/internal/streams/writable.js
index c8f1cd62861..5ee0cbc969f 100644
--- a/lib/internal/streams/writable.js
+++ b/lib/internal/streams/writable.js
@@ -693,24 +693,7 @@ function callFinal(stream, state) {
   state.pendingcb++;
 
   try {
-    const result = stream._final(onFinish);
-    if (result != null) {
-      const then = result.then;
-      if (typeof then === 'function') {
-        then.call(
-          result,
-          function() {
-            if (!called) {
-              process.nextTick(onFinish, null);
-            }
-          },
-          function(err) {
-            if (!called) {
-              process.nextTick(onFinish, err);
-            }
-          });
-      }
-    }
+    stream._final(onFinish);
   } catch (err) {
     onFinish(err);
   }