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

stream: add stream/iter Implementation · nodejs/node@28dc85d · GitHub

/ node Public

Commit 28dc85d

Browse files
jasnell
authored and
aduh95
committed
stream: add stream/iter Implementation
Experimental implementation of https://stream-iter.jasnell.me/ Signed-off-by: James M Snell <jasnell@gmail.com> Assisted-By: Claude/Opus 4.6 PR-URL: #62066 Backport-PR-URL: #64675 Reviewed-By: Robert Nagy <ronagy@icloud.com> Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
1 parent fbb3960 commit 28dc85d

27 files changed

Lines changed: 9426 additions & 302 deletions

‎doc/api/cli.md‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1203,6 +1203,16 @@ added:
12031203
12041204
Enable experimental support for storage inspection
12051205

1206+
### `--experimental-stream-iter`
1207+
1208+
<!-- YAML
1209+
added: REPLACEME
1210+
-->
1211+
1212+
> Stability: 1 - Experimental
1213+
1214+
Enable the experimental [`node:stream/iter`][] module.
1215+
12061216
### `--experimental-test-coverage`
12071217

12081218
<!-- YAML
@@ -3574,6 +3584,7 @@ one is included in the list below.
35743584
* `--experimental-require-module`
35753585
* `--experimental-shadow-realm`
35763586
* `--experimental-specifier-resolution`
3587+
* `--experimental-stream-iter`
35773588
* `--experimental-test-isolation`
35783589
* `--experimental-top-level-await`
35793590
* `--experimental-transform-types`
@@ -4212,6 +4223,7 @@ node --stack-trace-limit=12 -p -e "Error.stackTraceLimit" # prints 12
42124223
[`import` specifier]: esm.md#import-specifiers
42134224
[`net.getDefaultAutoSelectFamilyAttemptTimeout()`]: net.md#netgetdefaultautoselectfamilyattempttimeout
42144225
[`node:sqlite`]: sqlite.md
4226+
[`node:stream/iter`]: stream_iter.md
42154227
[`process.setUncaughtExceptionCaptureCallback()`]: process.md#processsetuncaughtexceptioncapturecallbackfn
42164228
[`tls.DEFAULT_MAX_VERSION`]: tls.md#tlsdefault_max_version
42174229
[`tls.DEFAULT_MIN_VERSION`]: tls.md#tlsdefault_min_version

‎doc/api/fs.md‎

Lines changed: 268 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -377,6 +377,154 @@ added: v10.0.0
377377
378378
* Type: {number} The numeric file descriptor managed by the {FileHandle} object.
379379
380+
#### `filehandle.pull([...transforms][, options])`
381+
382+
<!-- YAML
383+
added: REPLACEME
384+
-->
385+
386+
> Stability: 1 - Experimental
387+
388+
* `...transforms` {Function|Object} Optional transforms to apply via
389+
[`stream/iter pull()`][].
390+
* `options` {Object}
391+
* `signal` {AbortSignal}
392+
* `autoClose` {boolean} Close the file handle when the stream ends.
393+
**Default:** `false`.
394+
* `start` {number} Byte offset to begin reading from. When specified,
395+
reads use explicit positioning (`pread` semantics). **Default:** current
396+
file position.
397+
* `limit` {number} Maximum number of bytes to read before ending the
398+
iterator. Reads stop when `limit` bytes have been delivered or EOF is
399+
reached, whichever comes first. **Default:** read until EOF.
400+
* `chunkSize` {number} Size in bytes of the buffer allocated for each
401+
read operation. **Default:** `131072` (128 KB).
402+
* Returns: {AsyncIterable\<Uint8Array\[]>}
403+
404+
Return the file contents as an async iterable using the
405+
[`node:stream/iter`][] pull model. Reads are performed in `chunkSize`-byte
406+
chunks (default 128 KB). If transforms are provided, they are applied
407+
via [`stream/iter pull()`][].
408+
409+
The file handle is locked while the iterable is being consumed and unlocked
410+
when iteration completes, an error occurs, or the consumer breaks.
411+
412+
This function is only available when the `--experimental-stream-iter` flag is
413+
enabled.
414+
415+
```mjs
416+
import { open } from 'node:fs/promises';
417+
import { text } from 'node:stream/iter';
418+
import { compressGzip } from 'node:zlib/iter';
419+
420+
const fh = await open('input.txt', 'r');
421+
422+
// Read as text
423+
console.log(await text(fh.pull({ autoClose: true })));
424+
425+
// Read 1 KB starting at byte 100
426+
const fh2 = await open('input.txt', 'r');
427+
console.log(await text(fh2.pull({ start: 100, limit: 1024, autoClose: true })));
428+
429+
// Read with compression
430+
const fh3 = await open('input.txt', 'r');
431+
const compressed = fh3.pull(compressGzip(), { autoClose: true });
432+
```
433+
434+
```cjs
435+
const { open } = require('node:fs/promises');
436+
const { text } = require('node:stream/iter');
437+
const { compressGzip } = require('node:zlib/iter');
438+
439+
async function run() {
440+
const fh = await open('input.txt', 'r');
441+
442+
// Read as text
443+
console.log(await text(fh.pull({ autoClose: true })));
444+
445+
// Read 1 KB starting at byte 100
446+
const fh2 = await open('input.txt', 'r');
447+
console.log(await text(fh2.pull({ start: 100, limit: 1024, autoClose: true })));
448+
449+
// Read with compression
450+
const fh3 = await open('input.txt', 'r');
451+
const compressed = fh3.pull(compressGzip(), { autoClose: true });
452+
}
453+
454+
run().catch(console.error);
455+
```
456+
457+
#### `filehandle.pullSync([...transforms][, options])`
458+
459+
<!-- YAML
460+
added: REPLACEME
461+
-->
462+
463+
> Stability: 1 - Experimental
464+
465+
* `...transforms` {Function|Object} Optional transforms to apply via
466+
[`stream/iter pullSync()`][].
467+
* `options` {Object}
468+
* `autoClose` {boolean} Close the file handle when the stream ends.
469+
**Default:** `false`.
470+
* `start` {number} Byte offset to begin reading from. When specified,
471+
reads use explicit positioning. **Default:** current file position.
472+
* `limit` {number} Maximum number of bytes to read before ending the
473+
iterator. **Default:** read until EOF.
474+
* `chunkSize` {number} Size in bytes of the buffer allocated for each
475+
read operation. **Default:** `131072` (128 KB).
476+
* Returns: {Iterable\<Uint8Array\[]>}
477+
478+
Synchronous counterpart of [`filehandle.pull()`][]. Returns a sync iterable
479+
that reads the file using synchronous I/O on the main thread. Reads are
480+
performed in `chunkSize`-byte chunks (default 128 KB).
481+
482+
The file handle is locked while the iterable is being consumed. Unlike the
483+
async `pull()`, this method does not support `AbortSignal` since all
484+
operations are synchronous.
485+
486+
This function is only available when the `--experimental-stream-iter` flag is
487+
enabled.
488+
489+
```mjs
490+
import { open } from 'node:fs/promises';
491+
import { textSync, pipeToSync } from 'node:stream/iter';
492+
import { compressGzipSync, decompressGzipSync } from 'node:zlib/iter';
493+
494+
const fh = await open('input.txt', 'r');
495+
496+
// Read as text (sync)
497+
console.log(textSync(fh.pullSync({ autoClose: true })));
498+
499+
// Sync compress pipeline: file -> gzip -> file
500+
const src = await open('input.txt', 'r');
501+
const dst = await open('output.gz', 'w');
502+
pipeToSync(src.pullSync(compressGzipSync(), { autoClose: true }), dst.writer({ autoClose: true }));
503+
```
504+
505+
```cjs
506+
const { open } = require('node:fs/promises');
507+
const { textSync, pipeToSync } = require('node:stream/iter');
508+
const { compressGzipSync, decompressGzipSync } = require('node:zlib/iter');
509+
510+
async function run() {
511+
const fh = await open('input.txt', 'r');
512+
513+
// Read as text (sync)
514+
console.log(textSync(fh.pullSync({ autoClose: true })));
515+
516+
// Sync compress pipeline: file -> gzip -> file
517+
const src = await open('input.txt', 'r');
518+
const dst = await open('output.gz', 'w');
519+
pipeToSync(
520+
src.pullSync(compressGzipSync(), { autoClose: true }),
521+
dst.writer({ autoClose: true }),
522+
);
523+
}
524+
525+
run().catch(console.error);
526+
```
527+
380528
#### `filehandle.read(buffer, offset, length, position)`
381529
382530
<!-- YAML
@@ -905,6 +1053,121 @@ On Linux, positional writes don't work when the file is opened in append mode.
9051053
The kernel ignores the position argument and always appends the data to
9061054
the end of the file.
9071055
1056+
#### `filehandle.writer([options])`
1057+
1058+
<!-- YAML
1059+
added: REPLACEME
1060+
-->
1061+
1062+
> Stability: 1 - Experimental
1063+
1064+
* `options` {Object}
1065+
* `autoClose` {boolean} Close the file handle when the writer ends or
1066+
fails. **Default:** `false`.
1067+
* `start` {number} Byte offset to start writing at. When specified,
1068+
writes use explicit positioning. **Default:** current file position.
1069+
* `limit` {number} Maximum number of bytes the writer will accept.
1070+
Async writes (`write()`, `writev()`) that would exceed the limit reject
1071+
with `ERR_OUT_OF_RANGE`. Sync writes (`writeSync()`, `writevSync()`)
1072+
return `false`. **Default:** no limit.
1073+
* `chunkSize` {number} Maximum chunk size in bytes for synchronous write
1074+
operations. Writes larger than this threshold fall back to async I/O.
1075+
Set this to match the reader's `chunkSize` for optimal `pipeTo()`
1076+
performance. **Default:** `131072` (128 KB).
1077+
* Returns: {Object}
1078+
* `write(chunk[, options])` {Function} Returns {Promise\<void>}.
1079+
Accepts `Uint8Array`, `Buffer`, or string (UTF-8 encoded).
1080+
* `chunk` {Buffer|TypedArray|DataView|string}
1081+
* `options` {Object}
1082+
* `signal` {AbortSignal} If the signal is already aborted, the write
1083+
rejects with `AbortError` without performing I/O.
1084+
* `writev(chunks[, options])` {Function} Returns {Promise\<void>}. Uses
1085+
scatter/gather I/O via a single `writev()` syscall. Accepts mixed
1086+
`Uint8Array`/string arrays.
1087+
* `chunks` {Array\<Buffer|TypedArray|DataView|string>}
1088+
* `options` {Object}
1089+
* `signal` {AbortSignal} If the signal is already aborted, the write
1090+
rejects with `AbortError` without performing I/O.
1091+
* `writeSync(chunk)` {Function} Returns {boolean}. Attempts a synchronous
1092+
write. Returns `true` if the write succeeded, `false` if the caller
1093+
should fall back to async `write()`. Returns `false` when: the writer
1094+
is closed/errored, an async operation is in flight, the chunk exceeds
1095+
`chunkSize`, or the write would exceed `limit`.
1096+
* `chunk` {Buffer|TypedArray|DataView|string}
1097+
* `writevSync(chunks)` {Function} Returns {boolean}. Synchronous batch
1098+
write. Same fallback semantics as `writeSync()`.
1099+
* `chunks` {Array\<Buffer|TypedArray|DataView|string>}
1100+
* `end([options])` {Function} Returns {Promise\<number>} total bytes
1101+
written. Idempotent: returns `totalBytesWritten` if already closed,
1102+
returns the pending promise if already closing. Rejects if the writer
1103+
is in an errored state.
1104+
* `options` {Object}
1105+
* `signal` {AbortSignal} If the signal is already aborted, `end()`
1106+
rejects with `AbortError` and the writer remains open.
1107+
* `endSync()` {Function} Returns {number|number} total bytes written on
1108+
success, `-1` if the writer is errored or an async operation is in
1109+
flight. Idempotent when already closed.
1110+
* `fail(reason)` {Function} Puts the writer into a terminal error state.
1111+
Synchronous. If the writer is already closed or errored, this is a
1112+
no-op. If `autoClose` is true, closes the file handle synchronously.
1113+
1114+
Return a [`node:stream/iter`][] writer backed by this file handle.
1115+
1116+
The writer supports both `Symbol.asyncDispose` and `Symbol.dispose`:
1117+
1118+
* `await using w = fh.writer()` — if the writer is still open (no `end()`
1119+
called), `asyncDispose` calls `fail()`. If `end()` is pending, it waits
1120+
for it to complete.
1121+
* `using w = fh.writer()` — calls `fail()` unconditionally.
1122+
1123+
The `writeSync()` and `writevSync()` methods enable the try-sync fast path
1124+
used by [`stream/iter pipeTo()`][]. When the reader's chunk size matches the
1125+
writer's `chunkSize`, all writes in a `pipeTo()` pipeline complete
1126+
synchronously with zero promise overhead.
1127+
1128+
This function is only available when the `--experimental-stream-iter` flag is
1129+
enabled.
1130+
1131+
```mjs
1132+
import { open } from 'node:fs/promises';
1133+
import { from, pipeTo } from 'node:stream/iter';
1134+
import { compressGzip } from 'node:zlib/iter';
1135+
1136+
// Async pipeline
1137+
const fh = await open('output.gz', 'w');
1138+
await pipeTo(from('Hello!'), compressGzip(), fh.writer({ autoClose: true }));
1139+
1140+
// Sync pipeline with limit
1141+
const src = await open('input.txt', 'r');
1142+
const dst = await open('output.txt', 'w');
1143+
const w = dst.writer({ limit: 1024 * 1024 }); // Max 1 MB
1144+
await pipeTo(src.pull({ autoClose: true }), w);
1145+
await w.end();
1146+
await dst.close();
1147+
```
1148+
1149+
```cjs
1150+
const { open } = require('node:fs/promises');
1151+
const { from, pipeTo } = require('node:stream/iter');
1152+
const { compressGzip } = require('node:zlib/iter');
1153+
1154+
async function run() {
1155+
// Async pipeline
1156+
const fh = await open('output.gz', 'w');
1157+
await pipeTo(from('Hello!'), compressGzip(), fh.writer({ autoClose: true }));
1158+
1159+
// Sync pipeline with limit
1160+
const src = await open('input.txt', 'r');
1161+
const dst = await open('output.txt', 'w');
1162+
const w = dst.writer({ limit: 1024 * 1024 }); // Max 1 MB
1163+
await pipeTo(src.pull({ autoClose: true }), w);
1164+
await w.end();
1165+
await dst.close();
1166+
}
1167+
1168+
run().catch(console.error);
1169+
```
1170+
9081171
#### `filehandle[Symbol.asyncDispose]()`
9091172
9101173
<!-- YAML
@@ -8948,6 +9211,7 @@ the file contents.
89489211
[`event ports`]: https://illumos.org/man/port_create
89499212
[`filehandle.createReadStream()`]: #filehandlecreatereadstreamoptions
89509213
[`filehandle.createWriteStream()`]: #filehandlecreatewritestreamoptions
9214+
[`filehandle.pull()`]: #filehandlepulltransforms-options
89519215
[`filehandle.writeFile()`]: #filehandlewritefiledata-options
89529216
[`fs.access()`]: #fsaccesspath-mode-callback
89539217
[`fs.accessSync()`]: #fsaccesssyncpath-mode
@@ -8998,7 +9262,11 @@ the file contents.
89989262
[`inotify(7)`]: https://man7.org/linux/man-pages/man7/inotify.7.html
89999263
[`kqueue(2)`]: https://www.freebsd.org/cgi/man.cgi?query=kqueue&sektion=2
90009264
[`minimatch`]: https://github.com/isaacs/minimatch
9265+
[`node:stream/iter`]: stream_iter.md
90019266
[`statfs.bsize`]: #statfsbsize
9267+
[`stream/iter pipeTo()`]: stream_iter.md#pipetosource-transforms-writer
9268+
[`stream/iter pull()`]: stream_iter.md#pullsource-transforms-options
9269+
[`stream/iter pullSync()`]: stream_iter.md#pullsyncsource-transforms
90029270
[`util.promisify()`]: util.md#utilpromisifyoriginal
90039271
[bigints]: https://tc39.github.io/proposal-bigint
90049272
[caveats]: #caveats

‎doc/api/index.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
* [Modules: Packages](packages.md)
4444
* [Modules: TypeScript](typescript.md)
4545
* [Net](net.md)
46+
* [Iterable Streams API](stream_iter.md)
4647
* [OS](os.md)
4748
* [Path](path.md)
4849
* [Performance hooks](perf_hooks.md)
@@ -72,6 +73,7 @@
7273
* [Web Streams API](webstreams.md)
7374
* [Worker threads](worker_threads.md)
7475
* [Zlib](zlib.md)
76+
* [Zlib Iterable Compression](zlib_iter.md)
7577

7678
<hr class="line"/>
7779

0 commit comments

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