Skip to content

stream: add ReadableByteStream.tee() #44505

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nodejs-github-bot merged 2 commits into from
Sep 8, 2022
Merged

stream: add ReadableByteStream.tee() #44505

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2b7a1b3
stream: add `ReadableByteStream.tee()`
daeyeon Sep 3, 2022
af41bf0
fixup: update the doc
daeyeon Sep 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
stream: add ReadableByteStream.tee() 
This supports teeing readable byte streams to meet
the latest web streams standards.

Signed-off-by: Daeyeon Jeong daeyeon.dev@gmail.com
  • Loading branch information
@daeyeon
daeyeon committed Sep 3, 2022
commit 2b7a1b35a8e695c2d29921eb9fb201ccda32a0a5
304 changes: 293 additions & 11 deletions lib/internal/webstreams/readablestream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,7 @@ const {
ArrayBufferViewGetByteOffset,
ArrayBufferGetByteLength,
AsyncIterator,
cloneAsUint8Array,
copyArrayBuffer,
customInspect,
dequeueValue,
Expand Down Expand Up @@ -215,6 +216,7 @@ class ReadableStream {
throw new ERR_INVALID_ARG_VALUE('source', 'Object', source);
this[kState] = {
disturbed: false,
reader: undefined,
state: 'readable',
storedError: undefined,
stream: undefined,
Expand Down Expand Up @@ -1111,7 +1113,6 @@ class ReadableByteStreamController {
chunk);
}
const chunkByteLength = ArrayBufferViewGetByteLength(chunk);
const chunkByteOffset = ArrayBufferViewGetByteOffset(chunk);
const chunkBuffer = ArrayBufferViewGetBuffer(chunk);
const chunkBufferByteLength = ArrayBufferGetByteLength(chunkBuffer);
if (chunkByteLength === 0 || chunkBufferByteLength === 0) {
Expand All @@ -1122,11 +1123,7 @@ class ReadableByteStreamController {
throw new ERR_INVALID_STATE.TypeError('Controller is already closed');
if (this[kState].stream[kState].state !== 'readable')
throw new ERR_INVALID_STATE.TypeError('ReadableStream is already closed');
readableByteStreamControllerEnqueue(
this,
chunkBuffer,
chunkByteLength,
chunkByteOffset);
readableByteStreamControllerEnqueue(this, chunk);
}

/**
Expand Down Expand Up @@ -1430,6 +1427,13 @@ function readableStreamPipeTo(
}

function readableStreamTee(stream, cloneForBranch2) {
if (isReadableByteStreamController(stream[kState].controller)) {
return readableByteStreamTee(stream);
}
return readableStreamDefaultTee(stream, cloneForBranch2);
}

function readableStreamDefaultTee(stream, cloneForBranch2) {
const reader = new ReadableStreamDefaultReader(stream);
let reading = false;
let canceled1 = false;
Expand Down Expand Up @@ -1524,6 +1528,284 @@ function readableStreamTee(stream, cloneForBranch2) {
return [branch1, branch2];
}

function readableByteStreamTee(stream) {
assert(isReadableStream(stream));
assert(isReadableByteStreamController(stream[kState].controller));

let reader = new ReadableStreamDefaultReader(stream);
let reading = false;
let readAgainForBranch1 = false;
let readAgainForBranch2 = false;
let canceled1 = false;
let canceled2 = false;
let reason1;
let reason2;
let branch1;
let branch2;
const cancelDeferred = createDeferredPromise();

function forwardReaderError(thisReader) {
PromisePrototypeThen(
thisReader[kState].close.promise,
undefined,
(error) => {
if (thisReader !== reader) {
return;
}
readableStreamDefaultControllerError(branch1[kState].controller, error);
readableStreamDefaultControllerError(branch2[kState].controller, error);
if (!canceled1 || !canceled2) {
cancelDeferred.resolve();
}
}
);
}

function pullWithDefaultReader() {
if (isReadableStreamBYOBReader(reader)) {
readableStreamBYOBReaderRelease(reader);
reader = new ReadableStreamDefaultReader(stream);
forwardReaderError(reader);
}

const readRequest = {
[kChunk](chunk) {
queueMicrotask(() => {
readAgainForBranch1 = false;
readAgainForBranch2 = false;
const chunk1 = chunk;
let chunk2 = chunk;

if (!canceled1 && !canceled2) {
try {
chunk2 = cloneAsUint8Array(chunk);
} catch (error) {
readableByteStreamControllerError(
branch1[kState].controller,
error
);
readableByteStreamControllerError(
branch2[kState].controller,
error
);
cancelDeferred.resolve(readableStreamCancel(stream, error));
return;
}
}
if (!canceled1) {
readableByteStreamControllerEnqueue(
branch1[kState].controller,
chunk1
);
}
if (!canceled2) {
readableByteStreamControllerEnqueue(
branch2[kState].controller,
chunk2
);
}
reading = false;

if (readAgainForBranch1) {
pull1Algorithm();
} else if (readAgainForBranch2) {
pull2Algorithm();
}
});
},
[kClose]() {
reading = false;

if (!canceled1) {
readableByteStreamControllerClose(branch1[kState].controller);
}
if (!canceled2) {
readableByteStreamControllerClose(branch2[kState].controller);
}
if (branch1[kState].controller[kState].pendingPullIntos.length > 0) {
readableByteStreamControllerRespond(branch1[kState].controller, 0);
}
if (branch2[kState].controller[kState].pendingPullIntos.length > 0) {
readableByteStreamControllerRespond(branch2[kState].controller, 0);
}
if (!canceled1 || !canceled2) {
cancelDeferred.resolve();
}
},
[kError]() {
reading = false;
},
};

readableStreamDefaultReaderRead(reader, readRequest);
}

function pullWithBYOBReader(view, forBranch2) {
if (isReadableStreamDefaultReader(reader)) {
readableStreamDefaultReaderRelease(reader);
reader = new ReadableStreamBYOBReader(stream);
forwardReaderError(reader);
}

const byobBranch = forBranch2 === true ? branch2 : branch1;
const otherBranch = forBranch2 === false ? branch2 : branch1;
const readIntoRequest = {
[kChunk](chunk) {
queueMicrotask(() => {
readAgainForBranch1 = false;
readAgainForBranch2 = false;
const byobCanceled = forBranch2 === true ? canceled2 : canceled1;
const otherCanceled = forBranch2 === false ? canceled2 : canceled1;

if (!otherCanceled) {
let clonedChunk;

try {
clonedChunk = cloneAsUint8Array(chunk);
} catch (error) {
readableByteStreamControllerError(
byobBranch[kState].controller,
error
);
readableByteStreamControllerError(
otherBranch[kState].controller,
error
);
cancelDeferred.resolve(readableStreamCancel(stream, error));
return;
}
if (!byobCanceled) {
readableByteStreamControllerRespondWithNewView(
byobBranch[kState].controller,
chunk
);
}

readableByteStreamControllerEnqueue(
otherBranch[kState].controller,
clonedChunk
);
} else if (!byobCanceled) {
readableByteStreamControllerRespondWithNewView(
byobBranch[kState].controller,
chunk
);
}
reading = false;

if (readAgainForBranch1) {
pull1Algorithm();
} else if (readAgainForBranch2) {
pull2Algorithm();
}
});
},
[kClose](chunk) {
reading = false;

const byobCanceled = forBranch2 === true ? canceled2 : canceled1;
const otherCanceled = forBranch2 === false ? canceled2 : canceled1;

if (!byobCanceled) {
readableByteStreamControllerClose(byobBranch[kState].controller);
}
if (!otherCanceled) {
readableByteStreamControllerClose(otherBranch[kState].controller);
}
if (chunk !== undefined) {
if (!byobCanceled) {
readableByteStreamControllerRespondWithNewView(
byobBranch[kState].controller,
chunk
);
}
if (
!otherCanceled &&
otherBranch[kState].controller[kState].pendingPullIntos.length > 0
) {
readableByteStreamControllerRespond(
otherBranch[kState].controller,
0
);
}
}
if (!byobCanceled || !otherCanceled) {
cancelDeferred.resolve();
}
},
[kError]() {
reading = false;
},
};
readableStreamBYOBReaderRead(reader, view, readIntoRequest);
}

function pull1Algorithm() {
if (reading) {
readAgainForBranch1 = true;
return PromiseResolve();
}
reading = true;

const byobRequest = branch1[kState].controller.byobRequest;
if (byobRequest === null) {
pullWithDefaultReader();
} else {
pullWithBYOBReader(byobRequest[kState].view, false);
}
return PromiseResolve();
}

function pull2Algorithm() {
if (reading) {
readAgainForBranch2 = true;
return PromiseResolve();
}
reading = true;

const byobRequest = branch2[kState].controller.byobRequest;
if (byobRequest === null) {
pullWithDefaultReader();
} else {
pullWithBYOBReader(byobRequest[kState].view, true);
}
return PromiseResolve();
}

function cancel1Algorithm(reason) {
canceled1 = true;
reason1 = reason;
if (canceled2) {
cancelDeferred.resolve(readableStreamCancel(stream, [reason1, reason2]));
}
return cancelDeferred.promise;
}

function cancel2Algorithm(reason) {
canceled2 = true;
reason2 = reason;
if (canceled1) {
cancelDeferred.resolve(readableStreamCancel(stream, [reason1, reason2]));
}
return cancelDeferred.promise;
}

branch1 = new ReadableStream({
type: 'bytes',
pull: pull1Algorithm,
cancel: cancel1Algorithm,
});
branch2 = new ReadableStream({
type: 'bytes',
pull: pull2Algorithm,
cancel: cancel2Algorithm,
});

forwardReaderError(reader);

return [branch1, branch2];
}

function readableByteStreamControllerConvertPullIntoDescriptor(desc) {
const {
buffer,
Expand Down Expand Up @@ -2317,18 +2599,18 @@ function readableByteStreamControllerFillHeadPullIntoDescriptor(
desc.bytesFilled += size;
}

function readableByteStreamControllerEnqueue(
controller,
buffer,
byteLength,
byteOffset) {
function readableByteStreamControllerEnqueue(controller, chunk) {
const {
closeRequested,
pendingPullIntos,
queue,
stream,
} = controller[kState];

const buffer = ArrayBufferViewGetBuffer(chunk);
const byteOffset = ArrayBufferViewGetByteOffset(chunk);
const byteLength = ArrayBufferViewGetByteLength(chunk);

if (closeRequested || stream[kState].state !== 'readable')
return;

Expand Down
11 changes: 11 additions & 0 deletions lib/internal/webstreams/util.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

const {
ArrayBufferPrototype,
ArrayBufferPrototypeSlice,
ArrayPrototypePush,
ArrayPrototypeShift,
AsyncIteratorPrototype,
Expand Down Expand Up @@ -112,6 +113,15 @@ function ArrayBufferGetByteLength(view) {
return ReflectGet(ArrayBufferPrototype, 'byteLength', view);
}

function cloneAsUint8Array(view) {
const buffer = ArrayBufferViewGetBuffer(view);
const byteOffset = ArrayBufferViewGetByteOffset(view);
const byteLength = ArrayBufferViewGetByteLength(view);
return new Uint8Array(
ArrayBufferPrototypeSlice(buffer, byteOffset, byteOffset + byteLength)
);
}

function isBrandCheck(brand) {
return (value) => {
return value != null &&
Expand Down Expand Up @@ -236,6 +246,7 @@ module.exports = {
ArrayBufferViewGetByteOffset,
ArrayBufferGetByteLength,
AsyncIterator,
cloneAsUint8Array,
copyArrayBuffer,
customInspect,
dequeueValue,
Expand Down
Loading