| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent eee0e84 commit 193091f
23 files changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -183,22 +183,22 @@ The API supports two models: | |||
| 183 | 183 | Pull streams have natural backpressure -- the consumer drives the pace, so | |
| 184 | 184 | the source is never read faster than the consumer can process. Push streams | |
| 185 | 185 | need explicit backpressure because the producer and consumer run | |
| 186 | - independently. The `highWaterMark` and `backpressure` options on `push()`, | ||
| 186 | + independently. The `budget` and `backpressure` options on `push()`, | ||
| 187 | 187 | `broadcast()`, and `share()` control how this works. | |
| 188 | 188 | ||
| 189 | 189 | #### The two-buffer model | |
| 190 | 190 | ||
| 191 | 191 | Push streams use a two-part buffering system. Think of it like a bucket | |
| 192 | - (slots) being filled through a hose (pending writes), with a float valve | ||
| 192 | + (buffer) being filled through a hose (pending writes), with a float valve | ||
| 193 | 193 | that closes when the bucket is full: | |
| 194 | 194 | ||
| 195 | 195 | ```text | |
| 196 | - highWaterMark (e.g., 3) | ||
| 196 | + budget (e.g., 16384) | ||
| 197 | 197 | | | |
| 198 | 198 | Producer v | |
| 199 | 199 | | +---------+ | |
| 200 | 200 | v | | | |
| 201 | - [ write() ] ----+ +--->| slots |---> Consumer pulls | ||
| 201 | + [ write() ] ----+ +--->| buffer |---> Consumer pulls | ||
| 202 | 202 | [ write() ] | | | (bucket)| for await (...) | |
| 203 | 203 | [ write() ] v | +---------+ | |
| 204 | 204 | +--------+ ^ | |
@@ -211,29 +211,29 @@ that closes when the bucket is full: | |||
| 211 | 211 | 'strict' mode limits this too! | |
| 212 | 212 | ``` | |
| 213 | 213 | ||
| 214 | - * **Slots (the bucket)** -- data ready for the consumer, capped at | ||
| 215 | - `highWaterMark`. When the consumer pulls, it drains all slots at once | ||
| 216 | - into a single batch. | ||
| 214 | + * **Buffer (the bucket)** -- data ready for the consumer, capped at | ||
| 215 | + `budget` bytes. When the consumer pulls, it drains all buffered data | ||
| 216 | + at once into a single batch. | ||
| 217 | 217 | ||
| 218 | - * **Pending writes (the hose)** -- writes waiting for slot space. After | ||
| 218 | + * **Pending writes (the hose)** -- writes waiting for buffer space. After | ||
| 219 | 219 | the consumer drains, pending writes are promoted into the now-empty | |
| 220 | - slots and their promises settle. | ||
| 220 | + buffer and their promises settle. | ||
| 221 | 221 | ||
| 222 | 222 | How each policy uses these buffers: | |
| 223 | 223 | ||
| 224 | - | Policy | Slots limit | Pending writes limit | | ||
| 225 | - | --------------- | --------------- | -------------------- | | ||
| 226 | - | `'strict'` | `highWaterMark` | `highWaterMark` | | ||
| 227 | - | `'block'` | `highWaterMark` | Unbounded | | ||
| 228 | - | `'drop-oldest'` | `highWaterMark` | N/A (never waits) | | ||
| 229 | - | `'drop-newest'` | `highWaterMark` | N/A (never waits) | | ||
| 224 | + | Policy | Buffer limit | Pending writes limit | | ||
| 225 | + | --------------- | ------------ | -------------------- | | ||
| 226 | + | `'strict'` | `budget` | `budget` | | ||
| 227 | + | `'unbounded'` | `budget` | Unbounded | | ||
| 228 | + | `'drop-oldest'` | `budget` | N/A (never waits) | | ||
| 229 | + | `'drop-newest'` | `budget` | N/A (never waits) | | ||
| 230 | 230 | ||
| 231 | 231 | #### Strict (default) | |
| 232 | 232 | ||
| 233 | 233 | Strict mode catches "fire-and-forget" patterns where the producer calls | |
| 234 | 234 | `write()` without awaiting, which would cause unbounded memory growth. | |
| 235 | - It limits both the slots buffer and the pending writes queue to | ||
| 236 | - `highWaterMark`. | ||
| 235 | + It limits both the buffer and the pending writes queue to | ||
| 236 | + `budget` bytes. | ||
| 237 | 237 | ||
| 238 | 238 | If you properly await each write, you can only ever have one pending | |
| 239 | 239 | write at a time (yours), so you never hit the pending writes limit. | |
@@ -243,7 +243,7 @@ overflows: | |||
| 243 | 243 | ```mjs | |
| 244 | 244 | import { push, text } from 'node:stream/iter'; | |
| 245 | 245 | ||
| 246 | - const { writer, readable } = push({ highWaterMark: 16 }); | ||
| 246 | + const { writer, readable } = push({ budget: 16384 }); | ||
| 247 | 247 | ||
| 248 | 248 | // Consumer must run concurrently -- without it, the first write | |
| 249 | 249 | // that fills the buffer blocks the producer forever. | |
@@ -262,7 +262,7 @@ console.log(await consuming); | |||
| 262 | 262 | const { push, text } = require('node:stream/iter'); | |
| 263 | 263 | ||
| 264 | 264 | async function run() { | |
| 265 | - const { writer, readable } = push({ highWaterMark: 16 }); | ||
| 265 | + const { writer, readable } = push({ budget: 16384 }); | ||
| 266 | 266 | ||
| 267 | 267 | // Consumer must run concurrently -- without it, the first write | |
| 268 | 268 | // that fills the buffer blocks the producer forever. | |
@@ -290,9 +290,9 @@ for (const item of dataset) { | |||
| 290 | 290 | // --> throws "Backpressure violation: too many pending writes" | |
| 291 | 291 | ``` | |
| 292 | 292 | ||
| 293 | - #### Block | ||
| 293 | + #### Unbounded | ||
| 294 | 294 | ||
| 295 | - Block mode caps slots at `highWaterMark` but places no limit on the | ||
| 295 | + Unbounded mode caps buffered bytes at `budget` but places no limit on the | ||
| 296 | 296 | pending writes queue. Awaited writes block until the consumer makes room, | |
| 297 | 297 | just like strict mode. The difference is that unawaited writes silently | |
| 298 | 298 | queue forever instead of throwing -- a potential memory leak if the | |
@@ -306,8 +306,8 @@ properly, or when migrating code from those APIs. | |||
| 306 | 306 | import { push, text } from 'node:stream/iter'; | |
| 307 | 307 | ||
| 308 | 308 | const { writer, readable } = push({ | |
| 309 | - highWaterMark: 16, | ||
| 310 | - backpressure: 'block', | ||
| 309 | + budget: 16384, | ||
| 310 | + backpressure: 'unbounded', | ||
| 311 | 311 | }); | |
| 312 | 312 | ||
| 313 | 313 | const consuming = text(readable); | |
@@ -325,8 +325,8 @@ const { push, text } = require('node:stream/iter'); | |||
| 325 | 325 | ||
| 326 | 326 | async function run() { | |
| 327 | 327 | const { writer, readable } = push({ | |
| 328 | - highWaterMark: 16, | ||
| 329 | - backpressure: 'block', | ||
| 328 | + budget: 16384, | ||
| 329 | + backpressure: 'unbounded', | ||
| 330 | 330 | }); | |
| 331 | 331 | ||
| 332 | 332 | const consuming = text(readable); | |
@@ -352,19 +352,19 @@ any scenario where stale data is less valuable than current data. | |||
| 352 | 352 | ```mjs | |
| 353 | 353 | import { push } from 'node:stream/iter'; | |
| 354 | 354 | ||
| 355 | - // Keep only the 5 most recent readings | ||
| 355 | + // Keep only the most recent ~16 KB of readings | ||
| 356 | 356 | const { writer, readable } = push({ | |
| 357 | - highWaterMark: 5, | ||
| 357 | + budget: 16384, | ||
| 358 | 358 | backpressure: 'drop-oldest', | |
| 359 | 359 | }); | |
| 360 | 360 | ``` | |
| 361 | 361 | ||
| 362 | 362 | ```cjs | |
| 363 | 363 | const { push } = require('node:stream/iter'); | |
| 364 | 364 | ||
| 365 | - // Keep only the 5 most recent readings | ||
| 365 | + // Keep only the most recent ~16 KB of readings | ||
| 366 | 366 | const { writer, readable } = push({ | |
| 367 | - highWaterMark: 5, | ||
| 367 | + budget: 16384, | ||
| 368 | 368 | backpressure: 'drop-oldest', | |
| 369 | 369 | }); | |
| 370 | 370 | ``` | |
@@ -379,19 +379,19 @@ shedding load under pressure. | |||
| 379 | 379 | ```mjs | |
| 380 | 380 | import { push } from 'node:stream/iter'; | |
| 381 | 381 | ||
| 382 | - // Accept up to 10 buffered items; discard anything beyond that | ||
| 382 | + // Accept up to 16 KB of buffered data; discard anything beyond that | ||
| 383 | 383 | const { writer, readable } = push({ | |
| 384 | - highWaterMark: 10, | ||
| 384 | + budget: 16384, | ||
| 385 | 385 | backpressure: 'drop-newest', | |
| 386 | 386 | }); | |
| 387 | 387 | ``` | |
| 388 | 388 | ||
| 389 | 389 | ```cjs | |
| 390 | 390 | const { push } = require('node:stream/iter'); | |
| 391 | 391 | ||
| 392 | - // Accept up to 10 buffered items; discard anything beyond that | ||
| 392 | + // Accept up to 16 KB of buffered data; discard anything beyond that | ||
| 393 | 393 | const { writer, readable } = push({ | |
| 394 | - highWaterMark: 10, | ||
| 394 | + budget: 16384, | ||
| 395 | 395 | backpressure: 'drop-newest', | |
| 396 | 396 | }); | |
| 397 | 397 | ``` | |
@@ -413,14 +413,16 @@ if (writer.endSync() < 0) await writer.end(); | |||
| 413 | 413 | writer.fail(err); // Always synchronous, no fallback needed | |
| 414 | 414 | ``` | |
| 415 | 415 | ||
| 416 | - #### `writer.desiredSize` | ||
| 416 | + #### `writer.canWrite` | ||
| 417 | 417 | ||
| 418 | - * {number|null} | ||
| 418 | + * {boolean|null} | ||
| 419 | 419 | ||
| 420 | - The number of buffer slots available before the high water mark is reached. | ||
| 421 | - Returns `null` if the writer is closed or the consumer has disconnected. | ||
| 420 | + Returns `true` if the next write is likely to be accepted (buffered data is | ||
| 421 | + below capacity), `false` if backpressure is active, or `null` if the writer | ||
| 422 | + is closed or the consumer has disconnected. | ||
| 422 | 423 | ||
| 423 | - The value is always non-negative. | ||
| 424 | + This is a hint, not a guarantee: the state can change between the check and | ||
| 425 | + the write. Use [`ondrain()`][] to wait for capacity rather than polling. | ||
| 424 | 426 | ||
| 425 | 427 | #### `writer.end([options])` | |
| 426 | 428 | ||
@@ -756,10 +758,10 @@ added: REPLACEME | |||
| 756 | 758 | * `...transforms` {Function|Object} Optional transforms applied to the | |
| 757 | 759 | readable side. | |
| 758 | 760 | * `options` {Object} | |
| 759 | - * `highWaterMark` {number} Maximum number of buffered slots before | ||
| 760 | - backpressure is applied. Must be >= 1; values below 1 are clamped to 1. | ||
| 761 | - **Default:** `4`. | ||
| 762 | - * `backpressure` {string} Backpressure policy: `'strict'`, `'block'`, | ||
| 761 | + * `budget` {number} Maximum number of buffered bytes before | ||
| 762 | + backpressure is applied. Must be >= 16384. | ||
| 763 | + **Default:** `16384`. | ||
| 764 | + * `backpressure` {string} Backpressure policy: `'strict'`, `'unbounded'`, | ||
| 763 | 765 | `'drop-oldest'`, or `'drop-newest'`. **Default:** `'strict'`. | |
| 764 | 766 | * `signal` {AbortSignal} Abort the stream. | |
| 765 | 767 | * Returns: {Object} | |
@@ -818,18 +820,18 @@ added: REPLACEME | |||
| 818 | 820 | --> | |
| 819 | 821 | ||
| 820 | 822 | * `options` {Object} | |
| 821 | - * `highWaterMark` {number} Buffer size for both directions. | ||
| 822 | - **Default:** `4`. | ||
| 823 | + * `budget` {number} Buffer size in bytes for both directions. | ||
| 824 | + **Default:** `16384`. | ||
| 823 | 825 | * `backpressure` {string} Policy for both directions. | |
| 824 | 826 | **Default:** `'strict'`. | |
| 825 | 827 | * `signal` {AbortSignal} Cancellation signal for both channels. | |
| 826 | 828 | * `a` {Object} Options specific to the A-to-B direction. Overrides | |
| 827 | 829 | shared options. | |
| 828 | - * `highWaterMark` {number} | ||
| 830 | + * `budget` {number} | ||
| 829 | 831 | * `backpressure` {string} | |
| 830 | 832 | * `b` {Object} Options specific to the B-to-A direction. Overrides | |
| 831 | 833 | shared options. | |
| 832 | - * `highWaterMark` {number} | ||
| 834 | + * `budget` {number} | ||
| 833 | 835 | * `backpressure` {string} | |
| 834 | 836 | * Returns: {Array} A pair `[channelA, channelB]` of duplex channels. | |
| 835 | 837 | ||
@@ -1058,9 +1060,10 @@ fulfills with `true` when the writer can accept more data. | |||
| 1058 | 1060 | ```mjs | |
| 1059 | 1061 | import { push, ondrain, text } from 'node:stream/iter'; | |
| 1060 | 1062 | ||
| 1061 | - const { writer, readable } = push({ highWaterMark: 2 }); | ||
| 1062 | - writer.writeSync('a'); | ||
| 1063 | - writer.writeSync('b'); | ||
| 1063 | + const { writer, readable } = push({ budget: 16384 }); | ||
| 1064 | + const chunk = new Uint8Array(8192); // 8 KB | ||
| 1065 | + writer.writeSync(chunk); | ||
| 1066 | + writer.writeSync(chunk); // 16 KB total -- buffer full | ||
| 1064 | 1067 | ||
| 1065 | 1068 | // Start consuming so the buffer can actually drain | |
| 1066 | 1069 | const consuming = text(readable); | |
@@ -1078,9 +1081,10 @@ await consuming; | |||
| 1078 | 1081 | const { push, ondrain, text } = require('node:stream/iter'); | |
| 1079 | 1082 | ||
| 1080 | 1083 | async function run() { | |
| 1081 | - const { writer, readable } = push({ highWaterMark: 2 }); | ||
| 1082 | - writer.writeSync('a'); | ||
| 1083 | - writer.writeSync('b'); | ||
| 1084 | + const { writer, readable } = push({ budget: 16384 }); | ||
| 1085 | + const chunk = new Uint8Array(8192); // 8 KB | ||
| 1086 | + writer.writeSync(chunk); | ||
| 1087 | + writer.writeSync(chunk); // 16 KB total -- buffer full | ||
| 1084 | 1088 | ||
| 1085 | 1089 | // Start consuming so the buffer can actually drain | |
| 1086 | 1090 | const consuming = text(readable); | |
@@ -1189,9 +1193,9 @@ added: REPLACEME | |||
| 1189 | 1193 | --> | |
| 1190 | 1194 | ||
| 1191 | 1195 | * `options` {Object} | |
| 1192 | - * `highWaterMark` {number} Buffer size in slots. Must be >= 1; values | ||
| 1193 | - below 1 are clamped to 1. **Default:** `16`. | ||
| 1194 | - * `backpressure` {string} `'strict'`, `'block'`, `'drop-oldest'`, or | ||
| 1196 | + * `budget` {number} Buffer size in bytes. Must be >= 16384. | ||
| 1197 | + **Default:** `65536`. | ||
| 1198 | + * `backpressure` {string} `'strict'`, `'unbounded'`, `'drop-oldest'`, or | ||
| 1195 | 1199 | `'drop-newest'`. **Default:** `'strict'`. | |
| 1196 | 1200 | * `signal` {AbortSignal} | |
| 1197 | 1201 | * Returns: {Object} | |
@@ -1250,12 +1254,6 @@ async function run() { | |||
| 1250 | 1254 | run().catch(console.error); | |
| 1251 | 1255 | ``` | |
| 1252 | 1256 | ||
| 1253 | - #### `broadcast.bufferSize` | ||
| 1254 | - | ||
| 1255 | - * {number} | ||
| 1256 | - | ||
| 1257 | - The number of chunks currently buffered. | ||
| 1258 | - | ||
| 1259 | 1257 | #### `broadcast.cancel([reason])` | |
| 1260 | 1258 | ||
| 1261 | 1259 | * `reason` {Error} | |
@@ -1304,9 +1302,9 @@ added: REPLACEME | |||
| 1304 | 1302 | ||
| 1305 | 1303 | * `source` {AsyncIterable} The source to share. | |
| 1306 | 1304 | * `options` {Object} | |
| 1307 | - * `highWaterMark` {number} Buffer size. Must be >= 1; values below 1 | ||
| 1308 | - are clamped to 1. **Default:** `16`. | ||
| 1309 | - * `backpressure` {string} `'strict'`, `'block'`, `'drop-oldest'`, or | ||
| 1305 | + * `budget` {number} Buffer size in bytes. Must be >= 16384. | ||
| 1306 | + **Default:** `65536`. | ||
| 1307 | + * `backpressure` {string} `'strict'`, `'unbounded'`, `'drop-oldest'`, or | ||
| 1310 | 1308 | `'drop-newest'`. **Default:** `'strict'`. | |
| 1311 | 1309 | * Returns: {Share} | |
| 1312 | 1310 | ||
@@ -1544,7 +1542,7 @@ added: REPLACEME | |||
| 1544 | 1542 | * `backpressure` {string} Backpressure policy. **Default:** `'strict'`. | |
| 1545 | 1543 | * `'strict'` -- writes are rejected when the buffer is full. Catches | |
| 1546 | 1544 | callers that ignore backpressure. | |
| 1547 | - * `'block'` -- writes wait for drain when the buffer is full. Recommended | ||
| 1545 | + * `'unbounded'` -- writes wait for drain when the buffer is full. Recommended | ||
| 1548 | 1546 | for use with [`pipeTo()`][]. | |
| 1549 | 1547 | * `'drop-newest'` -- writes are silently discarded when the buffer is full. | |
| 1550 | 1548 | * `'drop-oldest'` -- **not supported**. Throws `ERR_INVALID_ARG_VALUE`. | |
@@ -1577,7 +1575,7 @@ const writable = new Writable({ | |||
| 1577 | 1575 | }); | |
| 1578 | 1576 | ||
| 1579 | 1577 | await pipeTo(from('hello world'), | |
| 1580 | - fromWritable(writable, { backpressure: 'block' })); | ||
| 1578 | + fromWritable(writable, { backpressure: 'unbounded' })); | ||
| 1581 | 1579 | ``` | |
| 1582 | 1580 | ||
| 1583 | 1581 | ```cjs | |
@@ -1590,7 +1588,7 @@ async function run() { | |||
| 1590 | 1588 | }); | |
| 1591 | 1589 | ||
| 1592 | 1590 | await pipeTo(from('hello world'), | |
| 1593 | - fromWritable(writable, { backpressure: 'block' })); | ||
| 1591 | + fromWritable(writable, { backpressure: 'unbounded' })); | ||
| 1594 | 1592 | } | |
| 1595 | 1593 | run(); | |
| 1596 | 1594 | ``` | |
@@ -2114,6 +2112,7 @@ console.log(textSync(stream)); // 'hello world' | |||
| 2114 | 2112 | [`from()`]: #frominput | |
| 2115 | 2113 | [`fromSync()`]: #fromsyncinput | |
| 2116 | 2114 | [`node:zlib/iter`]: zlib.md#iterable-compression | |
| 2115 | + [`ondrain()`]: #ondraindrainable | ||
| 2117 | 2116 | [`pipeTo()`]: #pipetosource-transforms-writer-options | |
| 2118 | 2117 | [`pull()`]: #pullsource-transforms-options | |
| 2119 | 2118 | [`pullSync()`]: #pullsyncsource-transforms | |
| Back | FazBrowse Home | New Git URL |
0 commit comments