FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

stream: handle enqueuing chunks when a pending BYOB pull request exists · nodejs/node@f300f19 · GitHub

/ node Public

Commit f300f19

Browse files
daeyeon
authored and
danielleadams
committed
stream: handle enqueuing chunks when a pending BYOB pull request exists
Signed-off-by: Daeyeon Jeong <daeyeon.dev@gmail.com> PR-URL: #44770 Refs: https://streams.spec.whatwg.org/#readable-byte-stream-controller-enqueue Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
1 parent 2a5bce6 commit f300f19

2 files changed

Lines changed: 58 additions & 25 deletions

File tree

‎lib/internal/webstreams/readablestream.js‎

Lines changed: 58 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -2647,13 +2647,22 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
26472647
);
26482648
}
26492649

2650-
firstPendingPullInto.buffer =
2651-
transferArrayBuffer(firstPendingPullInto.buffer);
2652-
}
2650+
readableByteStreamControllerInvalidateBYOBRequest(controller);
26532651

2654-
readableByteStreamControllerInvalidateBYOBRequest(controller);
2652+
firstPendingPullInto.buffer = transferArrayBuffer(
2653+
firstPendingPullInto.buffer
2654+
);
2655+
2656+
if (firstPendingPullInto.type === 'none') {
2657+
readableByteStreamControllerEnqueueDetachedPullIntoToQueue(
2658+
controller,
2659+
firstPendingPullInto
2660+
);
2661+
}
2662+
}
26552663

26562664
if (readableStreamHasDefaultReader(stream)) {
2665+
readableByteStreamControllerProcessReadRequestsUsingQueue(controller);
26572666
if (!readableStreamGetNumReadRequests(stream)) {
26582667
readableByteStreamControllerEnqueueChunkToQueue(
26592668
controller,
@@ -2662,6 +2671,10 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
26622671
byteLength);
26632672
} else {
26642673
assert(!queue.length);
2674+
if (pendingPullIntos.length) {
2675+
assert(pendingPullIntos[0].type === 'default');
2676+
readableByteStreamControllerShiftPendingPullInto(controller);
2677+
}
26652678
const transferredView =
26662679
new Uint8Array(transferredBuffer, byteOffset, byteLength);
26672680
readableStreamFulfillReadRequest(stream, transferredView, false);
@@ -2984,25 +2997,56 @@ function readableByteStreamControllerCancelSteps(controller, reason) {
29842997
return result;
29852998
}
29862999

3000+
function readableByteStreamControllerFillReadRequestFromQueue(controller, readRequest) {
3001+
const {
3002+
queue,
3003+
queueTotalSize,
3004+
} = controller[kState];
3005+
assert(queueTotalSize > 0);
3006+
const {
3007+
buffer,
3008+
byteOffset,
3009+
byteLength,
3010+
} = ArrayPrototypeShift(queue);
3011+
3012+
controller[kState].queueTotalSize -= byteLength;
3013+
readableByteStreamControllerHandleQueueDrain(controller);
3014+
const view = new Uint8Array(buffer, byteOffset, byteLength);
3015+
readRequest[kChunk](view);
3016+
}
3017+
3018+
function readableByteStreamControllerProcessReadRequestsUsingQueue(controller) {
3019+
const {
3020+
stream,
3021+
queueTotalSize,
3022+
} = controller[kState];
3023+
const { reader } = stream[kState];
3024+
assert(isReadableStreamDefaultReader(reader));
3025+
3026+
while (reader[kState].readRequests.length > 0) {
3027+
if (queueTotalSize === 0) {
3028+
return;
3029+
}
3030+
readableByteStreamControllerFillReadRequestFromQueue(
3031+
controller,
3032+
ArrayPrototypeShift(reader[kState].readRequests),
3033+
);
3034+
}
3035+
}
3036+
29873037
function readableByteStreamControllerPullSteps(controller, readRequest) {
29883038
const {
29893039
pendingPullIntos,
2990-
queue,
29913040
queueTotalSize,
29923041
stream,
29933042
} = controller[kState];
29943043
assert(readableStreamHasDefaultReader(stream));
29953044
if (queueTotalSize) {
29963045
assert(!readableStreamGetNumReadRequests(stream));
2997-
const {
2998-
buffer,
2999-
byteOffset,
3000-
byteLength,
3001-
} = ArrayPrototypeShift(queue);
3002-
controller[kState].queueTotalSize -= byteLength;
3003-
readableByteStreamControllerHandleQueueDrain(controller);
3004-
const view = new Uint8Array(buffer, byteOffset, byteLength);
3005-
readRequest[kChunk](view);
3046+
readableByteStreamControllerFillReadRequestFromQueue(
3047+
controller,
3048+
readRequest
3049+
);
30063050
return;
30073051
}
30083052
const {

‎test/wpt/status/streams.json‎

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -12,17 +12,6 @@
1212
]
1313
}
1414
},
15-
"readable-byte-streams/general.any.js": {
16-
"fail": {
17-
"expected": [
18-
"ReadableStream with byte source: enqueue() discards auto-allocated BYOB request",
19-
"ReadableStream with byte source: releaseLock() with pending read(view), read(view) on second reader, enqueue()",
20-
"ReadableStream with byte source: autoAllocateChunkSize, releaseLock() with pending read(), read() on second reader, enqueue()",
21-
"ReadableStream with byte source: autoAllocateChunkSize, releaseLock() with pending read(), read(view) on second reader, enqueue()",
22-
"ReadableStream with byte source: read(view) with 1 element Uint16Array, respond(1), releaseLock(), read() on second reader, enqueue()"
23-
]
24-
}
25-
},
2615
"readable-streams/cross-realm-crash.window.js": {
2716
"skip": "Browser-specific test"
2817
},

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL