FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

stream: expose DuplexPair API · nodejs/node@7c21bb9 · GitHub

/ node Public

Commit 7c21bb9

Browse files
awwright
authored and
marco-ippolito
committed
stream: expose DuplexPair API
PR-URL: #34111 Reviewed-By: Anna Henningsen <anna@addaleax.net> Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
1 parent 6a5120f commit 7c21bb9

32 files changed

Lines changed: 230 additions & 123 deletions

‎doc/api/stream.md‎

Lines changed: 30 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,8 +45,11 @@ There are four fundamental stream types within Node.js:
4545
is written and read (for example, [`zlib.createDeflate()`][]).
4646

4747
Additionally, this module includes the utility functions
48-
[`stream.pipeline()`][], [`stream.finished()`][], [`stream.Readable.from()`][]
49-
and [`stream.addAbortSignal()`][].
48+
[`stream.duplexPair()`][],
49+
[`stream.pipeline()`][],
50+
[`stream.finished()`][]
51+
[`stream.Readable.from()`][], and
52+
[`stream.addAbortSignal()`][].
5053

5154
### Streams Promises API
5255

@@ -2700,6 +2703,30 @@ unless `emitClose` is set in false.
27002703
Once `destroy()` has been called, any further calls will be a no-op and no
27012704
further errors except from `_destroy()` may be emitted as `'error'`.
27022705

2706+
#### `stream.duplexPair([options])`
2707+
2708+
<!-- YAML
2709+
added: REPLACEME
2710+
-->
2711+
2712+
* `options` {Object} A value to pass to both [`Duplex`][] constructors,
2713+
to set options such as buffering.
2714+
* Returns: {Array} of two [`Duplex`][] instances.
2715+
2716+
The utility function `duplexPair` returns an Array with two items,
2717+
each being a `Duplex` stream connected to the other side:
2718+
2719+
```js
2720+
const [ sideA, sideB ] = duplexPair();
2721+
```
2722+
2723+
Whatever is written to one stream is made readable on the other. It provides
2724+
behavior analogous to a network connection, where the data written by the client
2725+
becomes readable by the server, and vice-versa.
2726+
2727+
The Duplex streams are symmetrical; one or the other may be used without any
2728+
difference in behavior.
2729+
27032730
### `stream.finished(stream[, options], callback)`
27042731

27052732
<!-- YAML
@@ -4872,6 +4899,7 @@ contain multi-byte characters.
48724899
[`stream.addAbortSignal()`]: #streamaddabortsignalsignal-stream
48734900
[`stream.compose`]: #streamcomposestreams
48744901
[`stream.cork()`]: #writablecork
4902+
[`stream.duplexPair()`]: #streamduplexpairoptions
48754903
[`stream.finished()`]: #streamfinishedstream-options-callback
48764904
[`stream.pipe()`]: #readablepipedestination-options
48774905
[`stream.pipeline()`]: #streampipelinesource-transforms-destination-callback

‎lib/internal/streams/duplexpair.js‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
'use strict';
2+
const {
3+
Symbol,
4+
} = primordials;
5+
6+
const { Duplex } = require('stream');
7+
const assert = require('internal/assert');
8+
9+
const kCallback = Symbol('Callback');
10+
const kInitOtherSide = Symbol('InitOtherSide');
11+
12+
class DuplexSide extends Duplex {
13+
#otherSide = null;
14+
15+
constructor(options) {
16+
super(options);
17+
this[kCallback] = null;
18+
this.#otherSide = null;
19+
}
20+
21+
[kInitOtherSide](otherSide) {
22+
// Ensure this can only be set once, to enforce encapsulation.
23+
if (this.#otherSide === null) {
24+
this.#otherSide = otherSide;
25+
} else {
26+
assert(this.#otherSide === null);
27+
}
28+
}
29+
30+
_read() {
31+
const callback = this[kCallback];
32+
if (callback) {
33+
this[kCallback] = null;
34+
callback();
35+
}
36+
}
37+
38+
_write(chunk, encoding, callback) {
39+
assert(this.#otherSide !== null);
40+
assert(this.#otherSide[kCallback] === null);
41+
if (chunk.length === 0) {
42+
process.nextTick(callback);
43+
} else {
44+
this.#otherSide.push(chunk);
45+
this.#otherSide[kCallback] = callback;
46+
}
47+
}
48+
49+
_final(callback) {
50+
this.#otherSide.on('end', callback);
51+
this.#otherSide.push(null);
52+
}
53+
}
54+
55+
function duplexPair(options) {
56+
const side0 = new DuplexSide(options);
57+
const side1 = new DuplexSide(options);
58+
side0[kInitOtherSide](side1);
59+
side1[kInitOtherSide](side0);
60+
return [ side0, side1 ];
61+
}
62+
module.exports = duplexPair;

‎lib/stream.js‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,7 @@ Stream.Writable = require('internal/streams/writable');
101101
Stream.Duplex = require('internal/streams/duplex');
102102
Stream.Transform = require('internal/streams/transform');
103103
Stream.PassThrough = require('internal/streams/passthrough');
104+
Stream.duplexPair = require('internal/streams/duplexpair');
104105
Stream.pipeline = pipeline;
105106
const { addAbortSignal } = require('internal/streams/add-abort-signal');
106107
Stream.addAbortSignal = addAbortSignal;

‎test/common/README.md‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@ This directory contains modules used to test the Node.js implementation.
1212
* [CPU Profiler module](#cpu-profiler-module)
1313
* [Debugger module](#debugger-module)
1414
* [DNS module](#dns-module)
15-
* [Duplex pair helper](#duplex-pair-helper)
1615
* [Environment variables](#environment-variables)
1716
* [Fixtures module](#fixtures-module)
1817
* [Heap dump checker module](#heap-dump-checker-module)
@@ -669,14 +668,6 @@ Reads a Domain String and returns a Buffer containing the domain.
669668
Takes in a parsed Object and writes its fields to a DNS packet as a Buffer
670669
object.
671670

672-
## Duplex pair helper
673-
674-
The `common/duplexpair` module exports a single function `makeDuplexPair`,
675-
which returns an object `{ clientSide, serverSide }` where each side is a
676-
`Duplex` stream connected to the other side.
677-
678-
There is no difference between client or server side beyond their names.
679-
680671
## Environment variables
681672

682673
The behavior of the Node.js test suite can be altered using the following

‎test/common/duplexpair.js‎

Lines changed: 0 additions & 48 deletions
This file was deleted.

‎test/parallel/test-bootstrap-modules.js‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -126,6 +126,7 @@ if (common.isMainThread) {
126126
'NativeModule internal/streams/compose',
127127
'NativeModule internal/streams/destroy',
128128
'NativeModule internal/streams/duplex',
129+
'NativeModule internal/streams/duplexpair',
129130
'NativeModule internal/streams/end-of-stream',
130131
'NativeModule internal/streams/from',
131132
'NativeModule internal/streams/legacy',

‎test/parallel/test-gc-tls-external-memory.js‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ const common = require('../common');
88
if (!common.hasCrypto)
99
common.skip('missing crypto');
1010

11-
const makeDuplexPair = require('../common/duplexpair');
11+
const { duplexPair } = require('stream');
1212
const onGC = require('../common/ongc');
1313
const assert = require('assert');
1414
const tls = require('tls');
@@ -37,7 +37,7 @@ function connect() {
3737
return;
3838
}
3939

40-
const { clientSide, serverSide } = makeDuplexPair();
40+
const [ clientSide, serverSide ] = duplexPair();
4141

4242
const tlsSocket = tls.connect({ socket: clientSide });
4343
tlsSocket.on('error', common.mustCall(connect));

‎test/parallel/test-http-agent-domain-reused-gc.js‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
const common = require('../common');
44
const http = require('http');
55
const async_hooks = require('async_hooks');
6-
const makeDuplexPair = require('../common/duplexpair');
6+
const { duplexPair } = require('stream');
77

88
// Regression test for https://github.com/nodejs/node/issues/30122
99
// When a domain is attached to an http Agent’s ReusedHandle object, that
@@ -36,7 +36,7 @@ async_hooks.createHook({
3636
// attached to too many objects that use strong references (timers, the network
3737
// socket handle, etc.) and wrap the client side in a JSStreamSocket so we don’t
3838
// have to implement the whole _handle API ourselves.
39-
const { serverSide, clientSide } = makeDuplexPair();
39+
const [ serverSide, clientSide ] = duplexPair();
4040
const JSStreamSocket = require('internal/js_stream_socket');
4141
const wrappedClientSide = new JSStreamSocket(clientSide);
4242

‎test/parallel/test-http-generic-streams.js‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
const common = require('../common');
33
const assert = require('assert');
44
const http = require('http');
5-
const MakeDuplexPair = require('../common/duplexpair');
5+
const { duplexPair } = require('stream');
66

77
// Test 1: Simple HTTP test, no keep-alive.
88
{
@@ -13,7 +13,7 @@ const MakeDuplexPair = require('../common/duplexpair');
1313
res.end(testData);
1414
}));
1515

16-
const { clientSide, serverSide } = MakeDuplexPair();
16+
const [ clientSide, serverSide ] = duplexPair();
1717
server.emit('connection', serverSide);
1818

1919
const req = http.request({
@@ -37,7 +37,7 @@ const MakeDuplexPair = require('../common/duplexpair');
3737
res.end(testData);
3838
}, 2));
3939

40-
const { clientSide, serverSide } = MakeDuplexPair();
40+
const [ clientSide, serverSide ] = duplexPair();
4141
server.emit('connection', serverSide);
4242

4343
function doRequest(cb) {
@@ -77,7 +77,7 @@ const MakeDuplexPair = require('../common/duplexpair');
7777
});
7878
}));
7979

80-
const { clientSide, serverSide } = MakeDuplexPair();
80+
const [ clientSide, serverSide ] = duplexPair();
8181
server.emit('connection', serverSide);
8282
clientSide.on('end', common.mustCall());
8383
serverSide.on('end', common.mustCall());
@@ -117,7 +117,7 @@ const MakeDuplexPair = require('../common/duplexpair');
117117

118118
}));
119119

120-
const { clientSide, serverSide } = MakeDuplexPair();
120+
const [ clientSide, serverSide ] = duplexPair();
121121
server.emit('connection', serverSide);
122122
clientSide.on('end', common.mustCall());
123123
serverSide.on('end', common.mustCall());
@@ -143,7 +143,7 @@ const MakeDuplexPair = require('../common/duplexpair');
143143
{
144144
const server = http.createServer(common.mustNotCall());
145145

146-
const { clientSide, serverSide } = MakeDuplexPair();
146+
const [ clientSide, serverSide ] = duplexPair();
147147
server.emit('connection', serverSide);
148148

149149
server.on('clientError', common.mustCall());

‎test/parallel/test-http-insecure-parser-per-stream.js‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,13 @@
22
const common = require('../common');
33
const assert = require('assert');
44
const http = require('http');
5-
const MakeDuplexPair = require('../common/duplexpair');
5+
const { duplexPair } = require('stream');
66

77
// Test that setting the `maxHeaderSize` option works on a per-stream-basis.
88

99
// Test 1: The server sends an invalid header.
1010
{
11-
const { clientSide, serverSide } = MakeDuplexPair();
11+
const [ clientSide, serverSide ] = duplexPair();
1212

1313
const req = http.request({
1414
createConnection: common.mustCall(() => clientSide),
@@ -30,7 +30,7 @@ const MakeDuplexPair = require('../common/duplexpair');
3030

3131
// Test 2: The same as Test 1 except without the option, to make sure it fails.
3232
{
33-
const { clientSide, serverSide } = MakeDuplexPair();
33+
const [ clientSide, serverSide ] = duplexPair();
3434

3535
const req = http.request({
3636
createConnection: common.mustCall(() => clientSide)
@@ -59,7 +59,7 @@ const MakeDuplexPair = require('../common/duplexpair');
5959

6060
server.on('clientError', common.mustNotCall());
6161

62-
const { clientSide, serverSide } = MakeDuplexPair();
62+
const [ clientSide, serverSide ] = duplexPair();
6363
serverSide.server = server;
6464
server.emit('connection', serverSide);
6565

@@ -75,7 +75,7 @@ const MakeDuplexPair = require('../common/duplexpair');
7575

7676
server.on('clientError', common.mustCall());
7777

78-
const { clientSide, serverSide } = MakeDuplexPair();
78+
const [ clientSide, serverSide ] = duplexPair();
7979
serverSide.server = server;
8080
server.emit('connection', serverSide);
8181

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL