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>

6 files changed, 10 insertions, 360 deletions

diff --git a/doc/api/deprecations.md b/doc/api/deprecations.md
index b154c96c9eb..507ef307e60 100644
--- a/doc/api/deprecations.md
+++ b/doc/api/deprecations.md
@@ -3019,6 +3019,9 @@ it was an aborted or graceful destroy.
 
 <!-- YAML
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/40773
+    description: End-of-life.
   - version:
       - v17.2.0
       - v16.14.0
@@ -3026,7 +3029,7 @@ changes:
     description: Documentation-only deprecation.
 -->
 
-Type: Documentation-only
+Type: End-of-Life
 
 An undocumented feature of Node.js streams was to support thenables in
 implementation methods. This is now deprecated, use callbacks instead and avoid
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);
   }
diff --git a/test/parallel/test-stream-construct-async-error.js b/test/parallel/test-stream-construct-async-error.js
deleted file mode 100644
index 3fe81b4ebe2..00000000000
--- a/test/parallel/test-stream-construct-async-error.js
+++ /dev/null
@@ -1,232 +0,0 @@
-'use strict';
-
-const common = require('../common');
-const {
-  Duplex,
-  Writable,
-  Transform,
-} = require('stream');
-const { setTimeout } = require('timers/promises');
-const assert = require('assert');
-
-{
-  class Foo extends Duplex {
-    async _destroy(err, cb) {
-      await setTimeout(common.platformTimeout(1));
-      throw new Error('boom');
-    }
-  }
-
-  const foo = new Foo();
-  foo.destroy();
-  foo.on('error', common.expectsError({
-    message: 'boom'
-  }));
-  foo.on('close', common.mustCall(() => {
-    assert(foo.destroyed);
-  }));
-}
-
-{
-  class Foo extends Duplex {
-    async _destroy(err, cb) {
-      await setTimeout(common.platformTimeout(1));
-    }
-  }
-
-  const foo = new Foo();
-  foo.destroy();
-  foo.on('close', common.mustCall(() => {
-    assert(foo.destroyed);
-  }));
-}
-
-{
-  class Foo extends Duplex {
-    async _construct() {
-      await setTimeout(common.platformTimeout(1));
-    }
-
-    _write = common.mustCall((chunk, encoding, cb) => {
-      cb();
-    });
-
-    _read() {}
-  }
-
-  const foo = new Foo();
-  foo.write('test', common.mustCall());
-}
-
-{
-  class Foo extends Duplex {
-    async _construct(callback) {
-      await setTimeout(common.platformTimeout(1));
-      callback();
-    }
-
-    _write = common.mustCall((chunk, encoding, cb) => {
-      cb();
-    });
-
-    _read() {}
-  }
-
-  const foo = new Foo();
-  foo.write('test', common.mustCall());
-  foo.on('error', common.mustNotCall());
-}
-
-{
-  class Foo extends Writable {
-    _write = common.mustCall((chunk, encoding, cb) => {
-      cb();
-    });
-
-    async _final() {
-      await setTimeout(common.platformTimeout(1));
-    }
-  }
-
-  const foo = new Foo();
-  foo.end('hello');
-  foo.on('finish', common.mustCall());
-}
-
-{
-  class Foo extends Writable {
-    _write = common.mustCall((chunk, encoding, cb) => {
-      cb();
-    });
-
-    async _final(callback) {
-      await setTimeout(common.platformTimeout(1));
-      callback();
-    }
-  }
-
-  const foo = new Foo();
-  foo.end('hello');
-  foo.on('finish', common.mustCall());
-}
-
-{
-  class Foo extends Writable {
-    _write = common.mustCall((chunk, encoding, cb) => {
-      cb();
-    });
-
-    async _final() {
-      await setTimeout(common.platformTimeout(1));
-      throw new Error('boom');
-    }
-  }
-
-  const foo = new Foo();
-  foo.end('hello');
-  foo.on('error', common.expectsError({
-    message: 'boom'
-  }));
-  foo.on('close', common.mustCall());
-}
-
-{
-  const expected = ['hello', 'world'];
-  class Foo extends Transform {
-    async _flush() {
-      return 'world';
-    }
-
-    _transform(chunk, encoding, callback) {
-      callback(null, chunk);
-    }
-  }
-
-  const foo = new Foo();
-  foo.end('hello');
-  foo.on('data', common.mustCall((chunk) => {
-    assert.strictEqual(chunk.toString(), expected.shift());
-  }, 2));
-}
-
-{
-  const expected = ['hello', 'world'];
-  class Foo extends Transform {
-    async _flush(callback) {
-      callback(null, 'world');
-    }
-
-    _transform(chunk, encoding, callback) {
-      callback(null, chunk);
-    }
-  }
-
-  const foo = new Foo();
-  foo.end('hello');
-  foo.on('data', common.mustCall((chunk) => {
-    assert.strictEqual(chunk.toString(), expected.shift());
-  }, 2));
-}
-
-{
-  class Foo extends Transform {
-    async _flush(callback) {
-      throw new Error('boom');
-    }
-
-    _transform(chunk, encoding, callback) {
-      callback(null, chunk);
-    }
-  }
-
-  const foo = new Foo();
-  foo.end('hello');
-  foo.on('data', common.mustCall());
-  foo.on('error', common.expectsError({
-    message: 'boom'
-  }));
-  foo.on('close', common.mustCall());
-}
-
-{
-  class Foo extends Transform {
-    async _transform(chunk) {
-      return chunk.toString().toUpperCase();
-    }
-  }
-
-  const foo = new Foo();
-  foo.end('hello');
-  foo.on('data', common.mustCall((chunk) => {
-    assert.strictEqual(chunk.toString(), 'HELLO');
-  }));
-}
-
-{
-  class Foo extends Transform {
-    async _transform(chunk, _, callback) {
-      callback(null, chunk.toString().toUpperCase());
-    }
-  }
-
-  const foo = new Foo();
-  foo.end('hello');
-  foo.on('data', common.mustCall((chunk) => {
-    assert.strictEqual(chunk.toString(), 'HELLO');
-  }));
-}
-
-{
-  class Foo extends Transform {
-    async _transform() {
-      throw new Error('boom');
-    }
-  }
-
-  const foo = new Foo();
-  foo.end('hello');
-  foo.on('error', common.expectsError({
-    message: 'boom'
-  }));
-  foo.on('close', common.mustCall());
-}