From 15816bd0ddb0f1dcde21dcff3e412271d4464d10 Mon Sep 17 00:00:00 2001
From: Austin Wright <aaa@bzfx.net>
Date: Fri, 26 Jul 2024 01:09:23 -0700
Subject: [PATCH] stream: expose DuplexPair API

PR-URL: https://github.com/nodejs/node/pull/34111
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
---
 doc/api/stream.md                             | 32 +++++++-
 lib/internal/streams/duplexpair.js            | 62 ++++++++++++++++
 lib/stream.js                                 |  1 +
 test/common/README.md                         |  9 ---
 test/common/duplexpair.js                     | 48 ------------
 test/parallel/test-bootstrap-modules.js       |  1 +
 test/parallel/test-gc-tls-external-memory.js  |  4 +-
 .../test-http-agent-domain-reused-gc.js       |  4 +-
 test/parallel/test-http-generic-streams.js    | 12 +--
 .../test-http-insecure-parser-per-stream.js   | 10 +--
 .../test-http-max-header-size-per-stream.js   | 10 +--
 ...t-http-sync-write-error-during-continue.js |  4 +-
 test/parallel/test-http2-backpressure.js      |  4 +-
 .../test-http2-generic-streams-sendfile.js    |  4 +-
 test/parallel/test-http2-generic-streams.js   |  4 +-
 test/parallel/test-http2-padding-aligned.js   |  4 +-
 .../test-http2-perform-server-handshake.js    |  4 +-
 test/parallel/test-http2-sensitive-headers.js |  4 +-
 ...-http2-session-gc-while-write-scheduled.js |  4 +-
 test/parallel/test-http2-session-unref.js     |  4 +-
 ...tp2-write-finishes-after-stream-destroy.js |  4 +-
 .../test-https-insecure-parse-per-stream.js   |  7 +-
 .../test-https-max-header-size-per-stream.js  |  7 +-
 test/parallel/test-stream-duplexpair.js       | 74 +++++++++++++++++++
 test/parallel/test-tls-destroy-stream.js      |  4 +-
 test/parallel/test-tls-error-servername.js    |  4 +-
 test/parallel/test-tls-generic-stream.js      |  4 +-
 ...t-tls-socket-snicallback-without-server.js |  4 +-
 .../test-tls-streamwrap-buffersize.js         |  4 +-
 ...test-tls-transport-destroy-after-own-gc.js |  4 +-
 .../test-worker-http2-stream-terminate.js     |  4 +-
 ...orker-terminate-http2-respond-with-file.js |  4 +-
 32 files changed, 230 insertions(+), 123 deletions(-)
 create mode 100644 lib/internal/streams/duplexpair.js
 delete mode 100644 test/common/duplexpair.js
 create mode 100644 test/parallel/test-stream-duplexpair.js

diff --git a/doc/api/stream.md b/doc/api/stream.md
index c189df15c42957..1cf07d00093a26 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -45,8 +45,11 @@ There are four fundamental stream types within Node.js:
   is written and read (for example, [`zlib.createDeflate()`][]).
 
 Additionally, this module includes the utility functions
-[`stream.pipeline()`][], [`stream.finished()`][], [`stream.Readable.from()`][]
-and [`stream.addAbortSignal()`][].
+[`stream.duplexPair()`][],
+[`stream.pipeline()`][],
+[`stream.finished()`][]
+[`stream.Readable.from()`][], and
+[`stream.addAbortSignal()`][].
 
 ### Streams Promises API
 
@@ -2675,6 +2678,30 @@ unless `emitClose` is set in false.
 Once `destroy()` has been called, any further calls will be a no-op and no
 further errors except from `_destroy()` may be emitted as `'error'`.
 
+#### `stream.duplexPair([options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `options` {Object} A value to pass to both [`Duplex`][] constructors,
+  to set options such as buffering.
+* Returns: {Array} of two [`Duplex`][] instances.
+
+The utility function `duplexPair` returns an Array with two items,
+each being a `Duplex` stream connected to the other side:
+
+```js
+const [ sideA, sideB ] = duplexPair();
+```
+
+Whatever is written to one stream is made readable on the other. It provides
+behavior analogous to a network connection, where the data written by the client
+becomes readable by the server, and vice-versa.
+
+The Duplex streams are symmetrical; one or the other may be used without any
+difference in behavior.
+
 ### `stream.finished(stream[, options], callback)`
 
 <!-- YAML
@@ -4865,6 +4892,7 @@ contain multi-byte characters.
 [`stream.addAbortSignal()`]: #streamaddabortsignalsignal-stream
 [`stream.compose`]: #streamcomposestreams
 [`stream.cork()`]: #writablecork
+[`stream.duplexPair()`]: #streamduplexpairoptions
 [`stream.finished()`]: #streamfinishedstream-options-callback
 [`stream.pipe()`]: #readablepipedestination-options
 [`stream.pipeline()`]: #streampipelinesource-transforms-destination-callback
diff --git a/lib/internal/streams/duplexpair.js b/lib/internal/streams/duplexpair.js
new file mode 100644
index 00000000000000..a32084c4d4cbdf
--- /dev/null
+++ b/lib/internal/streams/duplexpair.js
@@ -0,0 +1,62 @@
+'use strict';
+const {
+  Symbol,
+} = primordials;
+
+const { Duplex } = require('stream');
+const assert = require('internal/assert');
+
+const kCallback = Symbol('Callback');
+const kInitOtherSide = Symbol('InitOtherSide');
+
+class DuplexSide extends Duplex {
+  #otherSide = null;
+
+  constructor(options) {
+    super(options);
+    this[kCallback] = null;
+    this.#otherSide = null;
+  }
+
+  [kInitOtherSide](otherSide) {
+    // Ensure this can only be set once, to enforce encapsulation.
+    if (this.#otherSide === null) {
+      this.#otherSide = otherSide;
+    } else {
+      assert(this.#otherSide === null);
+    }
+  }
+
+  _read() {
+    const callback = this[kCallback];
+    if (callback) {
+      this[kCallback] = null;
+      callback();
+    }
+  }
+
+  _write(chunk, encoding, callback) {
+    assert(this.#otherSide !== null);
+    assert(this.#otherSide[kCallback] === null);
+    if (chunk.length === 0) {
+      process.nextTick(callback);
+    } else {
+      this.#otherSide.push(chunk);
+      this.#otherSide[kCallback] = callback;
+    }
+  }
+
+  _final(callback) {
+    this.#otherSide.on('end', callback);
+    this.#otherSide.push(null);
+  }
+}
+
+function duplexPair(options) {
+  const side0 = new DuplexSide(options);
+  const side1 = new DuplexSide(options);
+  side0[kInitOtherSide](side1);
+  side1[kInitOtherSide](side0);
+  return [ side0, side1 ];
+}
+module.exports = duplexPair;
diff --git a/lib/stream.js b/lib/stream.js
index a69354138a2248..420415a2259827 100644
--- a/lib/stream.js
+++ b/lib/stream.js
@@ -101,6 +101,7 @@ Stream.Writable = require('internal/streams/writable');
 Stream.Duplex = require('internal/streams/duplex');
 Stream.Transform = require('internal/streams/transform');
 Stream.PassThrough = require('internal/streams/passthrough');
+Stream.duplexPair = require('internal/streams/duplexpair');
 Stream.pipeline = pipeline;
 const { addAbortSignal } = require('internal/streams/add-abort-signal');
 Stream.addAbortSignal = addAbortSignal;
diff --git a/test/common/README.md b/test/common/README.md
index a322242138740c..3e0dcd6ca73584 100644
--- a/test/common/README.md
+++ b/test/common/README.md
@@ -12,7 +12,6 @@ This directory contains modules used to test the Node.js implementation.
 * [CPU Profiler module](#cpu-profiler-module)
 * [Debugger module](#debugger-module)
 * [DNS module](#dns-module)
-* [Duplex pair helper](#duplex-pair-helper)
 * [Environment variables](#environment-variables)
 * [Fixtures module](#fixtures-module)
 * [Heap dump checker module](#heap-dump-checker-module)
@@ -669,14 +668,6 @@ Reads a Domain String and returns a Buffer containing the domain.
 Takes in a parsed Object and writes its fields to a DNS packet as a Buffer
 object.
 
-## Duplex pair helper
-
-The `common/duplexpair` module exports a single function `makeDuplexPair`,
-which returns an object `{ clientSide, serverSide }` where each side is a
-`Duplex` stream connected to the other side.
-
-There is no difference between client or server side beyond their names.
-
 ## Environment variables
 
 The behavior of the Node.js test suite can be altered using the following
diff --git a/test/common/duplexpair.js b/test/common/duplexpair.js
deleted file mode 100644
index 1f41ed32f1b9e9..00000000000000
--- a/test/common/duplexpair.js
+++ /dev/null
@@ -1,48 +0,0 @@
-'use strict';
-const { Duplex } = require('stream');
-const assert = require('assert');
-
-const kCallback = Symbol('Callback');
-const kOtherSide = Symbol('Other');
-
-class DuplexSocket extends Duplex {
-  constructor() {
-    super();
-    this[kCallback] = null;
-    this[kOtherSide] = null;
-  }
-
-  _read() {
-    const callback = this[kCallback];
-    if (callback) {
-      this[kCallback] = null;
-      callback();
-    }
-  }
-
-  _write(chunk, encoding, callback) {
-    assert.notStrictEqual(this[kOtherSide], null);
-    assert.strictEqual(this[kOtherSide][kCallback], null);
-    if (chunk.length === 0) {
-      process.nextTick(callback);
-    } else {
-      this[kOtherSide].push(chunk);
-      this[kOtherSide][kCallback] = callback;
-    }
-  }
-
-  _final(callback) {
-    this[kOtherSide].on('end', callback);
-    this[kOtherSide].push(null);
-  }
-}
-
-function makeDuplexPair() {
-  const clientSide = new DuplexSocket();
-  const serverSide = new DuplexSocket();
-  clientSide[kOtherSide] = serverSide;
-  serverSide[kOtherSide] = clientSide;
-  return { clientSide, serverSide };
-}
-
-module.exports = makeDuplexPair;
diff --git a/test/parallel/test-bootstrap-modules.js b/test/parallel/test-bootstrap-modules.js
index a8846913101e12..db59734207a7d8 100644
--- a/test/parallel/test-bootstrap-modules.js
+++ b/test/parallel/test-bootstrap-modules.js
@@ -128,6 +128,7 @@ if (common.isMainThread) {
     'NativeModule internal/streams/compose',
     'NativeModule internal/streams/destroy',
     'NativeModule internal/streams/duplex',
+    'NativeModule internal/streams/duplexpair',
     'NativeModule internal/streams/end-of-stream',
     'NativeModule internal/streams/from',
     'NativeModule internal/streams/legacy',
diff --git a/test/parallel/test-gc-tls-external-memory.js b/test/parallel/test-gc-tls-external-memory.js
index 2b584dd93dcf5b..752986019e0362 100644
--- a/test/parallel/test-gc-tls-external-memory.js
+++ b/test/parallel/test-gc-tls-external-memory.js
@@ -8,7 +8,7 @@ const common = require('../common');
 if (!common.hasCrypto)
   common.skip('missing crypto');
 
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 const onGC = require('../common/ongc');
 const assert = require('assert');
 const tls = require('tls');
@@ -37,7 +37,7 @@ function connect() {
     return;
   }
 
-  const { clientSide, serverSide } = makeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   const tlsSocket = tls.connect({ socket: clientSide });
   tlsSocket.on('error', common.mustCall(connect));
diff --git a/test/parallel/test-http-agent-domain-reused-gc.js b/test/parallel/test-http-agent-domain-reused-gc.js
index b63c2d20210c37..35146ee688eb9b 100644
--- a/test/parallel/test-http-agent-domain-reused-gc.js
+++ b/test/parallel/test-http-agent-domain-reused-gc.js
@@ -3,7 +3,7 @@
 const common = require('../common');
 const http = require('http');
 const async_hooks = require('async_hooks');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 // Regression test for https://github.com/nodejs/node/issues/30122
 // When a domain is attached to an http Agent’s ReusedHandle object, that
@@ -36,7 +36,7 @@ async_hooks.createHook({
 // attached to too many objects that use strong references (timers, the network
 // socket handle, etc.) and wrap the client side in a JSStreamSocket so we don’t
 // have to implement the whole _handle API ourselves.
-const { serverSide, clientSide } = makeDuplexPair();
+const [ serverSide, clientSide ] = duplexPair();
 const JSStreamSocket = require('internal/js_stream_socket');
 const wrappedClientSide = new JSStreamSocket(clientSide);
 
diff --git a/test/parallel/test-http-generic-streams.js b/test/parallel/test-http-generic-streams.js
index 706cba7b385957..1b2bc209c97121 100644
--- a/test/parallel/test-http-generic-streams.js
+++ b/test/parallel/test-http-generic-streams.js
@@ -2,7 +2,7 @@
 const common = require('../common');
 const assert = require('assert');
 const http = require('http');
-const MakeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 // Test 1: Simple HTTP test, no keep-alive.
 {
@@ -13,7 +13,7 @@ const MakeDuplexPair = require('../common/duplexpair');
     res.end(testData);
   }));
 
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
 
   const req = http.request({
@@ -37,7 +37,7 @@ const MakeDuplexPair = require('../common/duplexpair');
     res.end(testData);
   }, 2));
 
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
 
   function doRequest(cb) {
@@ -77,7 +77,7 @@ const MakeDuplexPair = require('../common/duplexpair');
     });
   }));
 
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
   clientSide.on('end', common.mustCall());
   serverSide.on('end', common.mustCall());
@@ -117,7 +117,7 @@ const MakeDuplexPair = require('../common/duplexpair');
 
   }));
 
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
   clientSide.on('end', common.mustCall());
   serverSide.on('end', common.mustCall());
@@ -143,7 +143,7 @@ const MakeDuplexPair = require('../common/duplexpair');
 {
   const server = http.createServer(common.mustNotCall());
 
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
 
   server.on('clientError', common.mustCall());
diff --git a/test/parallel/test-http-insecure-parser-per-stream.js b/test/parallel/test-http-insecure-parser-per-stream.js
index daee5811724eb6..5024b93af3d194 100644
--- a/test/parallel/test-http-insecure-parser-per-stream.js
+++ b/test/parallel/test-http-insecure-parser-per-stream.js
@@ -2,13 +2,13 @@
 const common = require('../common');
 const assert = require('assert');
 const http = require('http');
-const MakeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 // Test that setting the `maxHeaderSize` option works on a per-stream-basis.
 
 // Test 1: The server sends an invalid header.
 {
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   const req = http.request({
     createConnection: common.mustCall(() => clientSide),
@@ -30,7 +30,7 @@ const MakeDuplexPair = require('../common/duplexpair');
 
 // Test 2: The same as Test 1 except without the option, to make sure it fails.
 {
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   const req = http.request({
     createConnection: common.mustCall(() => clientSide)
@@ -59,7 +59,7 @@ const MakeDuplexPair = require('../common/duplexpair');
 
   server.on('clientError', common.mustNotCall());
 
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   serverSide.server = server;
   server.emit('connection', serverSide);
 
@@ -75,7 +75,7 @@ const MakeDuplexPair = require('../common/duplexpair');
 
   server.on('clientError', common.mustCall());
 
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   serverSide.server = server;
   server.emit('connection', serverSide);
 
diff --git a/test/parallel/test-http-max-header-size-per-stream.js b/test/parallel/test-http-max-header-size-per-stream.js
index c043fce21a8425..9ef794e718389e 100644
--- a/test/parallel/test-http-max-header-size-per-stream.js
+++ b/test/parallel/test-http-max-header-size-per-stream.js
@@ -2,13 +2,13 @@
 const common = require('../common');
 const assert = require('assert');
 const http = require('http');
-const MakeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 // Test that setting the `maxHeaderSize` option works on a per-stream-basis.
 
 // Test 1: The server sends larger headers than what would otherwise be allowed.
 {
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   const req = http.request({
     createConnection: common.mustCall(() => clientSide),
@@ -29,7 +29,7 @@ const MakeDuplexPair = require('../common/duplexpair');
 
 // Test 2: The same as Test 1 except without the option, to make sure it fails.
 {
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   const req = http.request({
     createConnection: common.mustCall(() => clientSide)
@@ -57,7 +57,7 @@ const MakeDuplexPair = require('../common/duplexpair');
 
   server.on('clientError', common.mustNotCall());
 
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   serverSide.server = server;
   server.emit('connection', serverSide);
 
@@ -73,7 +73,7 @@ const MakeDuplexPair = require('../common/duplexpair');
 
   server.on('clientError', common.mustCall());
 
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   serverSide.server = server;
   server.emit('connection', serverSide);
 
diff --git a/test/parallel/test-http-sync-write-error-during-continue.js b/test/parallel/test-http-sync-write-error-during-continue.js
index f77026b00b5131..87bbc515833c27 100644
--- a/test/parallel/test-http-sync-write-error-during-continue.js
+++ b/test/parallel/test-http-sync-write-error-during-continue.js
@@ -2,14 +2,14 @@
 const common = require('../common');
 const assert = require('assert');
 const http = require('http');
-const MakeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 // Regression test for the crash reported in
 // https://github.com/nodejs/node/issues/15102 (httpParser.finish() is called
 // during httpParser.execute()):
 
 {
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   serverSide.on('data', common.mustCall((data) => {
     assert.strictEqual(data.toString('utf8'), `\
diff --git a/test/parallel/test-http2-backpressure.js b/test/parallel/test-http2-backpressure.js
index 1f9687831eb1a8..40f147924407fe 100644
--- a/test/parallel/test-http2-backpressure.js
+++ b/test/parallel/test-http2-backpressure.js
@@ -7,7 +7,7 @@ if (!common.hasCrypto)
   common.skip('missing crypto');
 const assert = require('assert');
 const http2 = require('http2');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 {
   let req;
@@ -28,7 +28,7 @@ const makeDuplexPair = require('../common/duplexpair');
     assert.strictEqual(stream.write('A'.repeat(40)), false);
   }));
 
-  const { clientSide, serverSide } = makeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
 
   const client = http2.connect('http://localhost:80', {
diff --git a/test/parallel/test-http2-generic-streams-sendfile.js b/test/parallel/test-http2-generic-streams-sendfile.js
index b752b0fdcb815a..dab5175e93e744 100644
--- a/test/parallel/test-http2-generic-streams-sendfile.js
+++ b/test/parallel/test-http2-generic-streams-sendfile.js
@@ -5,7 +5,7 @@ if (!common.hasCrypto)
 const assert = require('assert');
 const http2 = require('http2');
 const fs = require('fs');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 {
   const server = http2.createServer();
@@ -13,7 +13,7 @@ const makeDuplexPair = require('../common/duplexpair');
     stream.respondWithFile(__filename);
   }));
 
-  const { clientSide, serverSide } = makeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
 
   const client = http2.connect('http://localhost:80', {
diff --git a/test/parallel/test-http2-generic-streams.js b/test/parallel/test-http2-generic-streams.js
index d97e86a5ecea55..85affcc8464290 100644
--- a/test/parallel/test-http2-generic-streams.js
+++ b/test/parallel/test-http2-generic-streams.js
@@ -4,7 +4,7 @@ if (!common.hasCrypto)
   common.skip('missing crypto');
 const assert = require('assert');
 const http2 = require('http2');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 {
   const testData = '<h1>Hello World</h1>';
@@ -17,7 +17,7 @@ const makeDuplexPair = require('../common/duplexpair');
     stream.end(testData);
   }));
 
-  const { clientSide, serverSide } = makeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
 
   const client = http2.connect('http://localhost:80', {
diff --git a/test/parallel/test-http2-padding-aligned.js b/test/parallel/test-http2-padding-aligned.js
index f687c02a98dc6e..0229985436185f 100644
--- a/test/parallel/test-http2-padding-aligned.js
+++ b/test/parallel/test-http2-padding-aligned.js
@@ -6,7 +6,7 @@ if (!common.hasCrypto)
 const assert = require('assert');
 const http2 = require('http2');
 const { PADDING_STRATEGY_ALIGNED, PADDING_STRATEGY_CALLBACK } = http2.constants;
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 {
   const testData = '<h1>Hello World.</h1>';
@@ -21,7 +21,7 @@ const makeDuplexPair = require('../common/duplexpair');
     stream.end(testData);
   }));
 
-  const { clientSide, serverSide } = makeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   // The lengths of the expected writes... note that this is highly
   // sensitive to how the internals are implemented.
diff --git a/test/parallel/test-http2-perform-server-handshake.js b/test/parallel/test-http2-perform-server-handshake.js
index b7fa67ec59d8fa..9cee5926238aab 100644
--- a/test/parallel/test-http2-perform-server-handshake.js
+++ b/test/parallel/test-http2-perform-server-handshake.js
@@ -8,11 +8,11 @@ if (!common.hasCrypto)
 const assert = require('assert');
 const http2 = require('http2');
 const stream = require('stream');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 // Basic test
 {
-  const { clientSide, serverSide } = makeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   const client = http2.connect('http://example.com', {
     createConnection: () => clientSide,
diff --git a/test/parallel/test-http2-sensitive-headers.js b/test/parallel/test-http2-sensitive-headers.js
index 7d4d775a55d4c2..aadedc1ba831b5 100644
--- a/test/parallel/test-http2-sensitive-headers.js
+++ b/test/parallel/test-http2-sensitive-headers.js
@@ -4,7 +4,7 @@ if (!common.hasCrypto)
   common.skip('missing crypto');
 const assert = require('assert');
 const http2 = require('http2');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 {
   const testData = '<h1>Hello World</h1>';
@@ -22,7 +22,7 @@ const makeDuplexPair = require('../common/duplexpair');
     stream.end(testData);
   }));
 
-  const { clientSide, serverSide } = makeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
 
   const client = http2.connect('http://localhost:80', {
diff --git a/test/parallel/test-http2-session-gc-while-write-scheduled.js b/test/parallel/test-http2-session-gc-while-write-scheduled.js
index aef9d9571d5559..62379f7d7ed678 100644
--- a/test/parallel/test-http2-session-gc-while-write-scheduled.js
+++ b/test/parallel/test-http2-session-gc-while-write-scheduled.js
@@ -5,7 +5,7 @@ const common = require('../common');
 if (!common.hasCrypto)
   common.skip('missing crypto');
 const http2 = require('http2');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 const tick = require('../common/tick');
 
 // This tests that running garbage collection while an Http2Session has
@@ -14,7 +14,7 @@ const tick = require('../common/tick');
 {
   // This creates a session and schedules a write (for the settings frame).
   let client = http2.connect('http://localhost:80', {
-    createConnection: common.mustCall(() => makeDuplexPair().clientSide)
+    createConnection: common.mustCall(() => duplexPair()[0])
   });
 
   // First, wait for any nextTicks() and their responses
diff --git a/test/parallel/test-http2-session-unref.js b/test/parallel/test-http2-session-unref.js
index 8afd659d33baa8..4b41d5cbc2a602 100644
--- a/test/parallel/test-http2-session-unref.js
+++ b/test/parallel/test-http2-session-unref.js
@@ -8,10 +8,10 @@ if (!common.hasCrypto)
   common.skip('missing crypto');
 const http2 = require('http2');
 const Countdown = require('../common/countdown');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 const server = http2.createServer();
-const { clientSide, serverSide } = makeDuplexPair();
+const [ clientSide, serverSide ] = duplexPair();
 
 const counter = new Countdown(3, () => server.unref());
 
diff --git a/test/parallel/test-http2-write-finishes-after-stream-destroy.js b/test/parallel/test-http2-write-finishes-after-stream-destroy.js
index 3b2dd4bcd4e548..ed8833fdb926b1 100644
--- a/test/parallel/test-http2-write-finishes-after-stream-destroy.js
+++ b/test/parallel/test-http2-write-finishes-after-stream-destroy.js
@@ -5,14 +5,14 @@ if (!common.hasCrypto)
   common.skip('missing crypto');
 const assert = require('assert');
 const http2 = require('http2');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
 // Make sure the Http2Stream destructor works, since we don't clean the
 // stream up like we would otherwise do.
 process.on('exit', global.gc);
 
 {
-  const { clientSide, serverSide } = makeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   let serverSideHttp2Stream;
   let serverSideHttp2StreamDestroyed = false;
diff --git a/test/parallel/test-https-insecure-parse-per-stream.js b/test/parallel/test-https-insecure-parse-per-stream.js
index 73e69d056913aa..5b21c9953067d6 100644
--- a/test/parallel/test-https-insecure-parse-per-stream.js
+++ b/test/parallel/test-https-insecure-parse-per-stream.js
@@ -7,9 +7,8 @@ if (!common.hasCrypto) {
 const fixtures = require('../common/fixtures');
 const assert = require('assert');
 const https = require('https');
-const MakeDuplexPair = require('../common/duplexpair');
 const tls = require('tls');
-const { finished } = require('stream');
+const { finished, duplexPair } = require('stream');
 
 const certFixture = {
   key: fixtures.readKey('agent1-key.pem'),
@@ -22,7 +21,7 @@ const certFixture = {
 
 // Test 1: The server sends an invalid header.
 {
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   const req = https.request({
     rejectUnauthorized: false,
@@ -45,7 +44,7 @@ const certFixture = {
 
 // Test 2: The same as Test 1 except without the option, to make sure it fails.
 {
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   const req = https.request({
     rejectUnauthorized: false,
diff --git a/test/parallel/test-https-max-header-size-per-stream.js b/test/parallel/test-https-max-header-size-per-stream.js
index f10fced777c2f4..3929169679ca48 100644
--- a/test/parallel/test-https-max-header-size-per-stream.js
+++ b/test/parallel/test-https-max-header-size-per-stream.js
@@ -10,8 +10,7 @@ const assert = require('assert');
 const https = require('https');
 const http = require('http');
 const tls = require('tls');
-const MakeDuplexPair = require('../common/duplexpair');
-const { finished } = require('stream');
+const { finished, duplexPair } = require('stream');
 
 const certFixture = {
   key: fixtures.readKey('agent1-key.pem'),
@@ -24,7 +23,7 @@ const certFixture = {
 
 // Test 1: The server sends larger headers than what would otherwise be allowed.
 {
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   const req = https.request({
     createConnection: common.mustCall(() => clientSide),
@@ -46,7 +45,7 @@ const certFixture = {
 
 // Test 2: The same as Test 1 except without the option, to make sure it fails.
 {
-  const { clientSide, serverSide } = MakeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
 
   const req = https.request({
     createConnection: common.mustCall(() => clientSide)
diff --git a/test/parallel/test-stream-duplexpair.js b/test/parallel/test-stream-duplexpair.js
new file mode 100644
index 00000000000000..3e1b3044ddb73c
--- /dev/null
+++ b/test/parallel/test-stream-duplexpair.js
@@ -0,0 +1,74 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { Duplex, duplexPair } = require('stream');
+
+{
+  const pair = duplexPair();
+
+  assert(pair[0] instanceof Duplex);
+  assert(pair[1] instanceof Duplex);
+  assert.notStrictEqual(pair[0], pair[1]);
+}
+
+{
+  // Verify that the iterable for array assignment works
+  const [ clientSide, serverSide ] = duplexPair();
+  assert(clientSide instanceof Duplex);
+  assert(serverSide instanceof Duplex);
+  clientSide.on(
+    'data',
+    common.mustCall((d) => assert.strictEqual(`${d}`, 'foo'))
+  );
+  clientSide.on('end', common.mustNotCall());
+  serverSide.write('foo');
+}
+
+{
+  const [ clientSide, serverSide ] = duplexPair();
+  assert(clientSide instanceof Duplex);
+  assert(serverSide instanceof Duplex);
+  serverSide.on(
+    'data',
+    common.mustCall((d) => assert.strictEqual(`${d}`, 'foo'))
+  );
+  serverSide.on('end', common.mustCall());
+  clientSide.end('foo');
+}
+
+{
+  const [ serverSide, clientSide ] = duplexPair();
+  serverSide.cork();
+  serverSide.write('abc');
+  serverSide.write('12');
+  serverSide.end('\n');
+  serverSide.uncork();
+  let characters = '';
+  clientSide.on('readable', function() {
+    for (let segment; (segment = this.read()) !== null;)
+      characters += segment;
+  });
+  clientSide.on('end', common.mustCall(function() {
+    assert.strictEqual(characters, 'abc12\n');
+  }));
+}
+
+// Test the case where the the _write never calls [kCallback]
+// because a zero-size push doesn't trigger a _read
+{
+  const [ serverSide, clientSide ] = duplexPair();
+  serverSide.write('');
+  serverSide.write('12');
+  serverSide.write('');
+  serverSide.write('');
+  serverSide.end('\n');
+  let characters = '';
+  clientSide.on('readable', function() {
+    for (let segment; (segment = this.read()) !== null;)
+      characters += segment;
+  });
+  clientSide.on('end', common.mustCall(function() {
+    assert.strictEqual(characters, '12\n');
+  }));
+}
diff --git a/test/parallel/test-tls-destroy-stream.js b/test/parallel/test-tls-destroy-stream.js
index a49e985a7e0204..7af85856093183 100644
--- a/test/parallel/test-tls-destroy-stream.js
+++ b/test/parallel/test-tls-destroy-stream.js
@@ -4,7 +4,7 @@ const common = require('../common');
 if (!common.hasCrypto) common.skip('missing crypto');
 
 const fixtures = require('../common/fixtures');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 const net = require('net');
 const assert = require('assert');
 const tls = require('tls');
@@ -41,7 +41,7 @@ const server = net.createServer((conn) => {
   conn.on('error', common.mustNotCall());
   // Assume that we want to use data to determine what to do with connections.
   conn.once('data', common.mustCall((chunk) => {
-    const { clientSide, serverSide } = makeDuplexPair();
+    const [ clientSide, serverSide ] = duplexPair();
     serverSide.on('close', common.mustCall(() => {
       conn.destroy();
     }));
diff --git a/test/parallel/test-tls-error-servername.js b/test/parallel/test-tls-error-servername.js
index 12e6f89e60fc0f..597b7f29a4064c 100644
--- a/test/parallel/test-tls-error-servername.js
+++ b/test/parallel/test-tls-error-servername.js
@@ -10,8 +10,8 @@ if (!common.hasCrypto)
 
 const assert = require('assert');
 const { connect, TLSSocket } = require('tls');
-const makeDuplexPair = require('../common/duplexpair');
-const { clientSide, serverSide } = makeDuplexPair();
+const { duplexPair } = require('stream');
+const [ clientSide, serverSide ] = duplexPair();
 
 const key = fixtures.readKey('agent1-key.pem');
 const cert = fixtures.readKey('agent1-cert.pem');
diff --git a/test/parallel/test-tls-generic-stream.js b/test/parallel/test-tls-generic-stream.js
index d4e5427acae110..8be72d7252718e 100644
--- a/test/parallel/test-tls-generic-stream.js
+++ b/test/parallel/test-tls-generic-stream.js
@@ -4,7 +4,7 @@ if (!common.hasCrypto)
   common.skip('missing crypto');
 
 const fixtures = require('../common/fixtures');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 const assert = require('assert');
 const { TLSSocket, connect } = require('tls');
 
@@ -12,7 +12,7 @@ const key = fixtures.readKey('agent1-key.pem');
 const cert = fixtures.readKey('agent1-cert.pem');
 const ca = fixtures.readKey('ca1-cert.pem');
 
-const { clientSide, serverSide } = makeDuplexPair();
+const [ clientSide, serverSide ] = duplexPair();
 
 const clientTLS = connect({
   socket: clientSide,
diff --git a/test/parallel/test-tls-socket-snicallback-without-server.js b/test/parallel/test-tls-socket-snicallback-without-server.js
index 3ef28b95702552..9ad5d554c81c39 100644
--- a/test/parallel/test-tls-socket-snicallback-without-server.js
+++ b/test/parallel/test-tls-socket-snicallback-without-server.js
@@ -10,9 +10,9 @@ if (!common.hasCrypto)
 const assert = require('assert');
 const tls = require('tls');
 const fixtures = require('../common/fixtures');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
-const { clientSide, serverSide } = makeDuplexPair();
+const [ clientSide, serverSide ] = duplexPair();
 new tls.TLSSocket(serverSide, {
   isServer: true,
   SNICallback: common.mustCall((servername, cb) => {
diff --git a/test/parallel/test-tls-streamwrap-buffersize.js b/test/parallel/test-tls-streamwrap-buffersize.js
index 0e0a49883a5a08..399c8b1c8cf272 100644
--- a/test/parallel/test-tls-streamwrap-buffersize.js
+++ b/test/parallel/test-tls-streamwrap-buffersize.js
@@ -4,7 +4,7 @@ if (!common.hasCrypto)
   common.skip('missing crypto');
 const assert = require('assert');
 const fixtures = require('../common/fixtures');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 const tls = require('tls');
 const net = require('net');
 
@@ -15,7 +15,7 @@ const net = require('net');
   const iter = 10;
 
   function createDuplex(port) {
-    const { clientSide, serverSide } = makeDuplexPair();
+    const [ clientSide, serverSide ] = duplexPair();
 
     return new Promise((resolve, reject) => {
       const socket = net.connect({
diff --git a/test/parallel/test-tls-transport-destroy-after-own-gc.js b/test/parallel/test-tls-transport-destroy-after-own-gc.js
index 819aa0a03d6cb4..17c494ca0b79d1 100644
--- a/test/parallel/test-tls-transport-destroy-after-own-gc.js
+++ b/test/parallel/test-tls-transport-destroy-after-own-gc.js
@@ -10,9 +10,9 @@ if (!common.hasCrypto)
   common.skip('missing crypto');
 
 const { TLSSocket } = require('tls');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 
-let { clientSide } = makeDuplexPair();
+let [ clientSide ] = duplexPair();
 
 let clientTLS = new TLSSocket(clientSide, { isServer: false });
 let clientTLSHandle = clientTLS._handle;  // eslint-disable-line no-unused-vars
diff --git a/test/parallel/test-worker-http2-stream-terminate.js b/test/parallel/test-worker-http2-stream-terminate.js
index 370961bec27923..128a79c3186d2e 100644
--- a/test/parallel/test-worker-http2-stream-terminate.js
+++ b/test/parallel/test-worker-http2-stream-terminate.js
@@ -4,7 +4,7 @@ if (!common.hasCrypto)
   common.skip('missing crypto');
 const assert = require('assert');
 const http2 = require('http2');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 const { parentPort, Worker } = require('worker_threads');
 
 // This test ensures that workers can be terminated without error while
@@ -46,7 +46,7 @@ if (!process.env.HAS_STARTED_WORKER) {
     stream.end('');
   });
 
-  const { clientSide, serverSide } = makeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
 
   const client = http2.connect('http://localhost:80', {
diff --git a/test/parallel/test-worker-terminate-http2-respond-with-file.js b/test/parallel/test-worker-terminate-http2-respond-with-file.js
index 8fa8602330fcfe..9c80d68a46c335 100644
--- a/test/parallel/test-worker-terminate-http2-respond-with-file.js
+++ b/test/parallel/test-worker-terminate-http2-respond-with-file.js
@@ -4,7 +4,7 @@ if (!common.hasCrypto)
   common.skip('missing crypto');
 const assert = require('assert');
 const http2 = require('http2');
-const makeDuplexPair = require('../common/duplexpair');
+const { duplexPair } = require('stream');
 const { Worker, isMainThread } = require('worker_threads');
 
 // This is a variant of test-http2-generic-streams-sendfile for checking
@@ -20,7 +20,7 @@ if (isMainThread) {
     stream.respondWithFile(process.execPath);  // Use a large-ish file.
   }));
 
-  const { clientSide, serverSide } = makeDuplexPair();
+  const [ clientSide, serverSide ] = duplexPair();
   server.emit('connection', serverSide);
 
   const client = http2.connect('http://localhost:80', {