mirror of
https://github.com/nodejs/node.git
synced 2026-10-28 18:25:13 +08:00
Ethan Arrowood
9e6dd49a12
Every consumer in node:stream/iter retains what it reads, so there is
no way to read a streamable to completion while keeping nothing. To
clear a stream, callers write `for await (const _ of source) {}`. This
is apparent throughout multiple test suites, including the stream/iter
tests themselves.
These are especially useful for QUIC where a receiver doesn't want the
payload, but still has to read it to completion to relieve
backpressure. bytes() does that too, but allocates the whole payload to
discard it.
dump() pulls every batch and drops it, so peak memory is one batch
regardless of volume. It takes the same signal and limit options as
the other consumers, rejects if the source errors mid-stream, and
fulfills with undefined. dumpSync() is the synchronous form.
The name follows undici's body.dump(). drain() was the obvious
alternative but this module already exports ondrain() and
drainableProtocol() for writer-side backpressure, which is the
opposite side of the pipe.
Assisted-by: Claude Opus 5
Signed-off-by: Ethan Arrowood <ethan@arrowood.dev>
PR-URL: https://github.com/nodejs/node/pull/65598
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Tim Perry <pimterry@gmail.com>
203 lines
3.2 KiB
JavaScript
203 lines
3.2 KiB
JavaScript
'use strict';
|
|
|
|
// Public entry point for the iterable streams API.
|
|
// Usage: require('stream/iter') or require('node:stream/iter')
|
|
// Requires: --experimental-stream-iter
|
|
|
|
const {
|
|
ObjectFreeze,
|
|
} = primordials;
|
|
|
|
const { emitExperimentalWarning } = require('internal/util');
|
|
emitExperimentalWarning('stream/iter');
|
|
|
|
// Protocol symbols
|
|
const {
|
|
toStreamable,
|
|
toAsyncStreamable,
|
|
broadcastProtocol,
|
|
shareProtocol,
|
|
shareSyncProtocol,
|
|
drainableProtocol,
|
|
} = require('internal/streams/iter/types');
|
|
|
|
// Factories
|
|
const { push } = require('internal/streams/iter/push');
|
|
const { duplex } = require('internal/streams/iter/duplex');
|
|
const { from, fromSync } = require('internal/streams/iter/from');
|
|
|
|
// Pipelines
|
|
const {
|
|
pull,
|
|
pullSync,
|
|
pipeTo,
|
|
pipeToSync,
|
|
} = require('internal/streams/iter/pull');
|
|
|
|
// Consumers
|
|
const {
|
|
bytes,
|
|
bytesSync,
|
|
text,
|
|
textSync,
|
|
arrayBuffer,
|
|
arrayBufferSync,
|
|
array,
|
|
arraySync,
|
|
dump,
|
|
dumpSync,
|
|
tap,
|
|
tapSync,
|
|
merge,
|
|
ondrain,
|
|
} = require('internal/streams/iter/consumers');
|
|
|
|
// Classic stream interop (Node.js-specific, not part of the spec)
|
|
const {
|
|
fromReadable,
|
|
fromWritable,
|
|
toReadable,
|
|
toReadableSync,
|
|
toWritable,
|
|
} = require('internal/streams/iter/classic');
|
|
|
|
// Multi-consumer
|
|
const { broadcast, Broadcast } = require('internal/streams/iter/broadcast');
|
|
const {
|
|
share,
|
|
shareSync,
|
|
Share,
|
|
SyncShare,
|
|
} = require('internal/streams/iter/share');
|
|
|
|
/**
|
|
* Stream namespace - unified access to all stream functions.
|
|
* @example
|
|
* const { Stream } = require('stream/iter');
|
|
*
|
|
* const { writer, readable } = Stream.push();
|
|
* await writer.write("hello");
|
|
* await writer.end();
|
|
*
|
|
* const output = Stream.pull(readable, transform1, transform2);
|
|
* const data = await Stream.bytes(output);
|
|
*/
|
|
const Stream = ObjectFreeze({
|
|
// Factories
|
|
push,
|
|
duplex,
|
|
from,
|
|
fromSync,
|
|
|
|
// Pipelines
|
|
pull,
|
|
pullSync,
|
|
|
|
// Pipe to destination
|
|
pipeTo,
|
|
pipeToSync,
|
|
|
|
// Consumers (async)
|
|
bytes,
|
|
text,
|
|
arrayBuffer,
|
|
array,
|
|
dump,
|
|
|
|
// Consumers (sync)
|
|
bytesSync,
|
|
textSync,
|
|
arrayBufferSync,
|
|
arraySync,
|
|
dumpSync,
|
|
|
|
// Combining
|
|
merge,
|
|
|
|
// Multi-consumer (push model)
|
|
broadcast,
|
|
|
|
// Multi-consumer (pull model)
|
|
share,
|
|
shareSync,
|
|
|
|
// Utilities
|
|
tap,
|
|
tapSync,
|
|
|
|
// Drain utility for event source integration
|
|
ondrain,
|
|
|
|
// Protocol symbols
|
|
toStreamable,
|
|
toAsyncStreamable,
|
|
broadcastProtocol,
|
|
shareProtocol,
|
|
shareSyncProtocol,
|
|
drainableProtocol,
|
|
});
|
|
|
|
module.exports = {
|
|
// The Stream namespace
|
|
Stream,
|
|
|
|
// Also export everything individually for destructured imports
|
|
|
|
// Protocol symbols
|
|
toStreamable,
|
|
toAsyncStreamable,
|
|
broadcastProtocol,
|
|
shareProtocol,
|
|
shareSyncProtocol,
|
|
drainableProtocol,
|
|
|
|
// Factories
|
|
push,
|
|
duplex,
|
|
from,
|
|
fromSync,
|
|
|
|
// Pipelines
|
|
pull,
|
|
pullSync,
|
|
pipeTo,
|
|
pipeToSync,
|
|
|
|
// Consumers (async)
|
|
bytes,
|
|
text,
|
|
arrayBuffer,
|
|
array,
|
|
dump,
|
|
|
|
// Consumers (sync)
|
|
bytesSync,
|
|
textSync,
|
|
arrayBufferSync,
|
|
arraySync,
|
|
dumpSync,
|
|
|
|
// Combining
|
|
merge,
|
|
|
|
// Multi-consumer
|
|
broadcast,
|
|
Broadcast,
|
|
share,
|
|
shareSync,
|
|
Share,
|
|
SyncShare,
|
|
|
|
// Utilities
|
|
tap,
|
|
tapSync,
|
|
ondrain,
|
|
|
|
// Classic stream interop
|
|
fromReadable,
|
|
fromWritable,
|
|
toReadable,
|
|
toReadableSync,
|
|
toWritable,
|
|
};
|