| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent eb72fe9 commit 9a45ff5
2 files changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -363,23 +363,37 @@ function toReadableSync(source, options = kNullPrototype) { | |||
| 363 | 363 | ||
| 364 | 364 | const ReadableCtor = lazyReadable(); | |
| 365 | 365 | const iterator = source[SymbolIterator](); | |
| 366 | + let hasBatch = false; | ||
| 367 | + let batch; | ||
| 368 | + let batchIndex = 0; | ||
| 366 | 369 | ||
| 367 | 370 | return new ReadableCtor({ | |
| 368 | 371 | __proto__: null, | |
| 369 | 372 | highWaterMark, | |
| 370 | 373 | read() { | |
| 371 | 374 | for (;;) { | |
| 372 | - const { value: batch, done } = iterator.next(); | ||
| 375 | + if (hasBatch) { | ||
| 376 | + while (batchIndex < batch.length) { | ||
| 377 | + if (!this.push(batch[batchIndex++])) return; | ||
| 378 | + } | ||
| 379 | + batch = undefined; | ||
| 380 | + hasBatch = false; | ||
| 381 | + batchIndex = 0; | ||
| 382 | + } | ||
| 383 | + | ||
| 384 | + const result = iterator.next(); | ||
| 385 | + const { done } = result; | ||
| 373 | 386 | if (done) { | |
| 374 | 387 | this.push(null); | |
| 375 | 388 | return; | |
| 376 | 389 | } | |
| 377 | - for (let i = 0; i < batch.length; i++) { | ||
| 378 | - if (!this.push(batch[i])) return; | ||
| 379 | - } | ||
| 390 | + batch = result.value; | ||
| 391 | + hasBatch = true; | ||
| 380 | 392 | } | |
| 381 | 393 | }, | |
| 382 | 394 | destroy(err, cb) { | |
| 395 | + batch = undefined; | ||
| 396 | + hasBatch = false; | ||
| 383 | 397 | if (typeof iterator.return === 'function') iterator.return(); | |
| 384 | 398 | cb(err); | |
| 385 | 399 | }, | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -439,6 +439,21 @@ async function testBackpressureSync() { | |||
| 439 | 439 | assert.strictEqual(chunks.length, 9); | |
| 440 | 440 | } | |
| 441 | 441 | ||
| 442 | + // ============================================================================= | ||
| 443 | + // fromStreamIterSync: backpressure within a batch | ||
| 444 | + // ============================================================================= | ||
| 445 | + | ||
| 446 | + async function testBackpressureSyncMultiChunkBatch() { | ||
| 447 | + function* gen() { | ||
| 448 | + yield [Buffer.from('a'), Buffer.from('b'), Buffer.from('c')]; | ||
| 449 | + } | ||
| 450 | + | ||
| 451 | + const readable = toReadableSync(gen(), { highWaterMark: 1 }); | ||
| 452 | + const result = await collect(readable); | ||
| 453 | + | ||
| 454 | + assert.strictEqual(result.toString(), 'abc'); | ||
| 455 | + } | ||
| 456 | + | ||
| 442 | 457 | // ============================================================================= | |
| 443 | 458 | // fromStreamIterSync: source error | |
| 444 | 459 | // ============================================================================= | |
@@ -613,6 +628,7 @@ Promise.all([ | |||
| 613 | 628 | testWithTransformAsync(), | |
| 614 | 629 | testBasicSync(), | |
| 615 | 630 | testBackpressureSync(), | |
| 631 | + testBackpressureSyncMultiChunkBatch(), | ||
| 616 | 632 | testErrorSync(), | |
| 617 | 633 | testDestroySync(), | |
| 618 | 634 | testRoundTrip(), | |
| Back | FazBrowse Home | New Git URL |
0 commit comments