| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent 6c53ddb commit aa6913c
3 files changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -1517,8 +1517,9 @@ the synchronous Writer methods (`writeSync`, `writevSync`, `endSync`) always | |||
| 1517 | 1517 | return `false` or `-1`, deferring to the async path. The per-write | |
| 1518 | 1518 | `options.signal` parameter from the Writer interface is also ignored. | |
| 1519 | 1519 | ||
| 1520 | - The result is cached per instance -- calling `fromWritable()` twice with the | ||
| 1521 | - same stream returns the same Writer. | ||
| 1520 | + The result is cached per instance and backpressure policy -- calling | ||
| 1521 | + `fromWritable()` twice with the same stream and `backpressure` option returns | ||
| 1522 | + the same Writer. | ||
| 1522 | 1523 | ||
| 1523 | 1524 | For duck-typed streams that do not expose `writableHighWaterMark`, | |
| 1524 | 1525 | `writableLength`, or similar properties, sensible defaults are used. | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -21,6 +21,7 @@ const { | |||
| 21 | 21 | PromiseReject, | |
| 22 | 22 | PromiseResolve, | |
| 23 | 23 | PromiseWithResolvers, | |
| 24 | + SafeMap, | ||
| 24 | 25 | SafeWeakMap, | |
| 25 | 26 | SymbolAsyncDispose, | |
| 26 | 27 | SymbolAsyncIterator, | |
@@ -427,10 +428,6 @@ function fromWritable(writable, options = kNullPrototype) { | |||
| 427 | 428 | throw new ERR_INVALID_ARG_TYPE('writable', 'Writable', writable); | |
| 428 | 429 | } | |
| 429 | 430 | ||
| 430 | - // Return cached adapter if available. | ||
| 431 | - const cached = fromWritableCache.get(writable); | ||
| 432 | - if (cached !== undefined) return cached; | ||
| 433 | - | ||
| 434 | 431 | validateObject(options, 'options'); | |
| 435 | 432 | const { | |
| 436 | 433 | backpressure = 'strict', | |
@@ -459,6 +456,17 @@ function fromWritable(writable, options = kNullPrototype) { | |||
| 459 | 456 | 'drop-oldest is not supported for classic stream.Writable'); | |
| 460 | 457 | } | |
| 461 | 458 | ||
| 459 | + // Return cached adapter if available. Backpressure policy changes writer | ||
| 460 | + // behavior, so cache one adapter per policy. | ||
| 461 | + let cachedByBackpressure = fromWritableCache.get(writable); | ||
| 462 | + if (cachedByBackpressure !== undefined) { | ||
| 463 | + const cached = cachedByBackpressure.get(backpressure); | ||
| 464 | + if (cached !== undefined) return cached; | ||
| 465 | + } else { | ||
| 466 | + cachedByBackpressure = new SafeMap(); | ||
| 467 | + fromWritableCache.set(writable, cachedByBackpressure); | ||
| 468 | + } | ||
| 469 | + | ||
| 462 | 470 | // Fall back to sensible defaults for duck-typed streams that may not | |
| 463 | 471 | // expose the full stream.Writable property set. | |
| 464 | 472 | const hwm = writable.writableHighWaterMark ?? 16384; | |
@@ -710,7 +718,7 @@ function fromWritable(writable, options = kNullPrototype) { | |||
| 710 | 718 | return promise; | |
| 711 | 719 | }; | |
| 712 | 720 | ||
| 713 | - fromWritableCache.set(writable, writer); | ||
| 721 | + cachedByBackpressure.set(backpressure, writer); | ||
| 714 | 722 | return writer; | |
| 715 | 723 | } | |
| 716 | 724 | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -0,0 +1,45 @@ | |||
| 1 | + // Flags: --experimental-stream-iter | ||
| 2 | + 'use strict'; | ||
| 3 | + | ||
| 4 | + const common = require('../common'); | ||
| 5 | + const assert = require('assert'); | ||
| 6 | + const { Writable } = require('stream'); | ||
| 7 | + const { fromWritable } = require('stream/iter'); | ||
| 8 | + | ||
| 9 | + { | ||
| 10 | + const writable = new Writable({ write() {} }); | ||
| 11 | + | ||
| 12 | + fromWritable(writable); | ||
| 13 | + | ||
| 14 | + assert.throws( | ||
| 15 | + () => fromWritable(writable, { backpressure: 'invalid' }), | ||
| 16 | + { code: 'ERR_INVALID_ARG_VALUE' }, | ||
| 17 | + ); | ||
| 18 | + | ||
| 19 | + writable.destroy(); | ||
| 20 | + } | ||
| 21 | + | ||
| 22 | + async function testCachedWritableUsesLaterBackpressureOptions() { | ||
| 23 | + const chunks = []; | ||
| 24 | + const writable = new Writable({ | ||
| 25 | + highWaterMark: 1, | ||
| 26 | + write(chunk, encoding, callback) { | ||
| 27 | + chunks.push(Buffer.from(chunk)); | ||
| 28 | + }, | ||
| 29 | + }); | ||
| 30 | + | ||
| 31 | + fromWritable(writable); | ||
| 32 | + const writer = fromWritable(writable, { backpressure: 'drop-newest' }); | ||
| 33 | + | ||
| 34 | + await writer.write('a'); | ||
| 35 | + await writer.write('b'); | ||
| 36 | + | ||
| 37 | + assert.deepStrictEqual( | ||
| 38 | + chunks.map((chunk) => chunk.toString()), | ||
| 39 | + ['a'], | ||
| 40 | + ); | ||
| 41 | + | ||
| 42 | + writable.destroy(); | ||
| 43 | + } | ||
| 44 | + | ||
| 45 | + testCachedWritableUsesLaterBackpressureOptions().then(common.mustCall()); | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments