Skip to content

Commit 0e1c99a

Browse files
jasnellrichardlau
authored andcommitted
doc: document stream/iter behaviors and Node.js extensions
Document behaviors that were previously only visible in the code: - the writer "closing" state after end()/endSync(), - writes made from argument conversion being ordered first, - zero-length push() writes not being buffered, - broadcast.cancel() also closing the paired writer, - merge() not waiting for the other sources' cleanup on error, - the options argument passed to the tap() callback, - FileHandle writer() ordering of un-awaited writes, and the lazy locking of FileHandle pull() and pullSync(). Also link the WinterTC Iterable Streams API draft and list the exports that are Node.js extensions to it. Assisted-by: OpenCode Signed-off-by: James M Snell <[email protected]> PR-URL: #66483 Reviewed-By: Trivikram Kamat <[email protected]>
1 parent 1683353 commit 0e1c99a

2 files changed

Lines changed: 55 additions & 7 deletions

File tree

‎doc/api/fs.md‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -410,8 +410,9 @@ Return the file contents as an async iterable using the
410410
chunks (default 128 KB). If transforms are provided, they are applied
411411
via [`stream/iter pull()`][].
412412
413-
The file handle is locked while the iterable is being consumed and unlocked
414-
when iteration completes, an error occurs, or the consumer breaks.
413+
The file handle is locked from the first read of the iterable, and unlocked
414+
when iteration completes, an error occurs, or the consumer breaks. An iterable
415+
that is never read does not lock the file handle.
415416
416417
This function is only available when the `--experimental-stream-iter` flag is
417418
enabled.
@@ -483,8 +484,8 @@ Synchronous counterpart of [`filehandle.pull()`][]. Returns a sync iterable
483484
that reads the file using synchronous I/O on the main thread. Reads are
484485
performed in `chunkSize`-byte chunks (default 128 KB).
485486
486-
The file handle is locked while the iterable is being consumed. Unlike the
487-
async `pull()`, this method does not support `AbortSignal` since all
487+
The file handle is locked from the first read of the iterable until iteration
488+
ends, as with [`filehandle.pull()`][]. Unlike the async `pull()`, this method does not support `AbortSignal` since all
488489
operations are synchronous.
489490
490491
This function is only available when the `--experimental-stream-iter` flag is
@@ -1124,6 +1125,11 @@ The writer supports both `Symbol.asyncDispose` and `Symbol.dispose`:
11241125
for it to complete.
11251126
* `using w = fh.writer()` — calls `fail()` unconditionally.
11261127
1128+
Async writes (`write()` and `writev()`) that are started without awaiting the
1129+
previous one are performed one at a time, in the order they were called, so
1130+
they never overlap in the file. A queued write is not performed if the writer
1131+
fails, or its `signal` aborts, before its turn.
1132+
11271133
The `writeSync()` and `writevSync()` methods enable the try-sync fast path
11281134
used by [`stream/iter pipeTo()`][]. When the reader's chunk size matches the
11291135
writer's `chunkSize`, all writes in a `pipeTo()` pipeline complete

‎doc/api/stream_iter.md‎

Lines changed: 45 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,13 @@ functions or objects with a `transform` method.
1818
Data flows in **batches** ({Uint8Array\[]} per iteration) to amortize the cost
1919
of async operations.
2020

21+
The module implements the WinterTC [Iterable Streams API][] draft. The
22+
classic stream interop functions ([`fromReadable()`][], [`fromWritable()`][],
23+
[`toReadable()`][], [`toReadableSync()`][] and [`toWritable()`][]),
24+
[`Broadcast.from()`][], [`Share.from()`][], [`SyncShare.fromSync()`][] and the
25+
protocol symbols exported by `Stream` are Node.js extensions that are not part
26+
of the draft.
27+
2128
```mjs
2229
import { from, pull, text } from 'node:stream/iter';
2330
import { compressGzip, decompressGzip } from 'node:zlib/iter';
@@ -407,6 +414,18 @@ converted to a `USVString` and then UTF-8 encoded. `writev()` and
407414
chunks. Writer option dictionaries treat `null` as an empty dictionary and
408415
ignore unknown members.
409416

417+
Arguments are converted before the write itself starts. If the conversion runs
418+
user code (for example a `toString()` method, or the iterator of a `writev()`
419+
argument) that writes to the same writer, those writes are ordered before the
420+
write whose argument is being converted, and they count against the same
421+
backpressure limits.
422+
423+
After `end()` or `endSync()` has been called, and until all buffered data has
424+
been consumed, the writer is _closing_. While closing, `canWrite` is `null`,
425+
`write()` and `writev()` reject with a `TypeError`, `writeSync()` and
426+
`writevSync()` return `false`, `endSync()` returns `-1`, and calling `end()`
427+
again returns the same promise as the first call.
428+
410429
Each async method has a synchronous `*Sync` counterpart designed for a
411430
try-fallback pattern: attempt the fast synchronous path first, and fall back
412431
to the async version only when the synchronous call indicates it could not
@@ -855,6 +874,10 @@ run().catch(console.error);
855874

856875
The writer returned by `push()` conforms to the \[Writer interface]\[].
857876

877+
Zero-length chunks are accepted without being buffered: they are not delivered
878+
to the consumer, and `writeSync()` and `write()` report success for them even
879+
when backpressure is active.
880+
858881
## Duplex channels
859882

860883
### `duplex([options])`
@@ -1239,7 +1262,12 @@ added: v25.9.0
12391262

12401263
Merge multiple async iterables by yielding batches in temporal order
12411264
(whichever source produces data first). All sources are consumed
1242-
concurrently.
1265+
concurrently, with at most one pending `next()` call per source.
1266+
1267+
If a source fails, the returned iterable rejects with its error. `merge()`
1268+
calls `return()` on the other sources but does not wait for it to settle: an
1269+
async generator source that is suspended in an `await` only runs its cleanup
1270+
once that `await` completes.
12431271

12441272
```mjs
12451273
import { from, merge, text } from 'node:stream/iter';
@@ -1265,8 +1293,9 @@ run().catch(console.error);
12651293
added: v25.9.0
12661294
-->
12671295

1268-
* `callback` {Function} `(chunks) => void` Called with each batch and with
1269-
`null` when the source ends.
1296+
* `callback` {Function} `(chunks, options) => void` Called with each batch and
1297+
with `null` when the source ends. `options.signal` is the pipeline's
1298+
{AbortSignal}.
12701299
* Returns: {Function} A stateless transform.
12711300

12721301
Create a pass-through transform that observes batches without modifying them.
@@ -1391,6 +1420,10 @@ run().catch(console.error);
13911420
Cancel the broadcast. If `reason` is provided, all consumers reject with that
13921421
exact reason. If it is omitted, consumers complete normally.
13931422

1423+
Cancelling also closes the paired writer: afterwards its `canWrite` is `null`
1424+
and `write()` rejects with a `TypeError`. This lets a [`Broadcast.from()`][]
1425+
pump stop pulling from its source.
1426+
13941427
#### `broadcast.consumerCount`
13951428

13961429
* {number}
@@ -2282,13 +2315,19 @@ const stream = fromSync(new Greeting('world'));
22822315
console.log(textSync(stream)); // 'hello world'
22832316
```
22842317
2318+
[Iterable Streams API]: https://iter-streams.proposal.wintertc.org/
22852319
[`--experimental-stream-iter`]: cli.md#--experimental-stream-iter
2320+
[`Broadcast.from()`]: #broadcastfrominput-options
2321+
[`Share.from()`]: #static-method-sharefrominput-options
2322+
[`SyncShare.fromSync()`]: #static-method-syncsharefromsyncinput-options
22862323
[`array()`]: #arraysource-options
22872324
[`arrayBuffer()`]: #arraybuffersource-options
22882325
[`bytes()`]: #bytessource-options
22892326
[`dump()`]: #drainsource-options
22902327
[`from()`]: #frominput
2328+
[`fromReadable()`]: #fromreadablereadable
22912329
[`fromSync()`]: #fromsyncinput
2330+
[`fromWritable()`]: #fromwritablewritable-options
22922331
[`node:zlib/iter`]: zlib.md#iterable-compression
22932332
[`ondrain()`]: #ondraindrainable
22942333
[`pipeTo()`]: #pipetosource-transforms-writer-options
@@ -2300,3 +2339,6 @@ console.log(textSync(stream)); // 'hello world'
23002339
[`tap()`]: #tapcallback
23012340
[`text()`]: #textsource-options
23022341
[`toAsyncStreamable`]: #streamtoasyncstreamable
2342+
[`toReadable()`]: #toreadablesource-options
2343+
[`toReadableSync()`]: #toreadablesyncsource-options
2344+
[`toWritable()`]: #towritablewriter

0 commit comments

Comments
 (0)