| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent 2f749dd commit e264c5c
2 files changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -1512,6 +1512,8 @@ void Stream::EndWriting() { | |||
| 1512 | 1512 | void Stream::EntryRead(size_t amount) { | |
| 1513 | 1513 | // Called when the JS consumer reads data from the inbound DataQueue. | |
| 1514 | 1514 | // Extend the flow control window so the sender can transmit more. | |
| 1515 | + if (session().is_destroyed()) return; | ||
| 1516 | + Session::SendPendingDataScope send_scope(&session()); | ||
| 1515 | 1517 | session().ExtendStreamOffset(id(), amount); | |
| 1516 | 1518 | session().ExtendOffset(amount); | |
| 1517 | 1519 | } | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -0,0 +1,62 @@ | |||
| 1 | + // Flags: --experimental-quic --experimental-stream-iter --no-warnings | ||
| 2 | + | ||
| 3 | + // Test: a sender blocked on flow control is resumed once the consumer | ||
| 4 | + // starts reading. This confirms that we do flush MAX_STREAM_DATA | ||
| 5 | + // frames as expected when the consumer reads. | ||
| 6 | + | ||
| 7 | + import { hasQuic, skip, mustCall, mustCallAtLeast } from '../common/index.mjs'; | ||
| 8 | + import { setTimeout as delay } from 'node:timers/promises'; | ||
| 9 | + import assert from 'node:assert'; | ||
| 10 | + | ||
| 11 | + const { strictEqual, deepStrictEqual } = assert; | ||
| 12 | + | ||
| 13 | + if (!hasQuic) { | ||
| 14 | + skip('QUIC is not enabled'); | ||
| 15 | + } | ||
| 16 | + | ||
| 17 | + const { listen, connect } = await import('../common/quic.mjs'); | ||
| 18 | + const { bytes } = await import('stream/iter'); | ||
| 19 | + | ||
| 20 | + // Larger than the default 256 KB stream flow-control window: | ||
| 21 | + const size = 1024 * 1024; | ||
| 22 | + const expected = new Uint8Array(size); | ||
| 23 | + for (let i = 0; i < size; i++) expected[i] = i & 0xff; | ||
| 24 | + | ||
| 25 | + const senderBlocked = Promise.withResolvers(); | ||
| 26 | + const done = Promise.withResolvers(); | ||
| 27 | + | ||
| 28 | + const serverEndpoint = await listen(mustCall((serverSession) => { | ||
| 29 | + serverSession.onstream = mustCall(async (stream) => { | ||
| 30 | + stream.onblocked = mustCallAtLeast(() => senderBlocked.resolve(stream), 1); | ||
| 31 | + stream.setBody(expected); | ||
| 32 | + await stream.closed; | ||
| 33 | + }); | ||
| 34 | + })); | ||
| 35 | + | ||
| 36 | + const clientSession = await connect(serverEndpoint.address); | ||
| 37 | + await clientSession.opened; | ||
| 38 | + | ||
| 39 | + // Write a byte to open the stream: | ||
| 40 | + const stream = await clientSession.createBidirectionalStream(); | ||
| 41 | + await stream.writer.write(new Uint8Array([1])); | ||
| 42 | + | ||
| 43 | + // Wait until the sender has filled the window and blocked. | ||
| 44 | + const serverStream = await senderBlocked.promise; | ||
| 45 | + | ||
| 46 | + // Poll until everything has been acked, so the stream is fully idle: | ||
| 47 | + while (serverStream.stats.maxOffsetAcknowledged !== serverStream.stats.bytesSent) { | ||
| 48 | + await delay(1); | ||
| 49 | + } | ||
| 50 | + | ||
| 51 | + // Try to read: | ||
| 52 | + const received = await bytes(stream); | ||
| 53 | + strictEqual(received.byteLength, expected.byteLength); | ||
| 54 | + deepStrictEqual(received, expected); | ||
| 55 | + | ||
| 56 | + stream.writer.endSync(); | ||
| 57 | + await stream.closed; | ||
| 58 | + clientSession.close(); | ||
| 59 | + done.resolve(); | ||
| 60 | + | ||
| 61 | + await Promise.all([done.promise, clientSession.closed]); | ||
| 62 | + await serverEndpoint.close(); | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments