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

stream: fixes for webstreams · nodejs/node@48138af · GitHub

/ node Public

Commit 48138af

Browse files
MattiasBuelens
authored and
marco-ippolito
committed
stream: fixes for webstreams
PR-URL: #51168 Backport-PR-URL: #52773 Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
1 parent 1e2bc2c commit 48138af

5 files changed

Lines changed: 249 additions & 280 deletions

File tree

‎lib/internal/webstreams/readablestream.js‎

Lines changed: 86 additions & 92 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@ const {
1111
FunctionPrototypeCall,
1212
MathMin,
1313
NumberIsInteger,
14-
ObjectCreate,
1514
ObjectDefineProperties,
1615
ObjectSetPrototypeOf,
1716
Promise,
@@ -96,9 +95,9 @@ const {
9695
AsyncIterator,
9796
cloneAsUint8Array,
9897
copyArrayBuffer,
98+
createPromiseCallback,
9999
customInspect,
100100
dequeueValue,
101-
ensureIsPromise,
102101
enqueueValueWithSize,
103102
extractHighWaterMark,
104103
extractSizeAlgorithm,
@@ -251,19 +250,7 @@ class ReadableStream {
251250
constructor(source = {}, strategy = kEmptyObject) {
252251
if (source === null)
253252
throw new ERR_INVALID_ARG_VALUE('source', 'Object', source);
254-
this[kState] = {
255-
disturbed: false,
256-
reader: undefined,
257-
state: 'readable',
258-
storedError: undefined,
259-
stream: undefined,
260-
transfer: {
261-
writable: undefined,
262-
port1: undefined,
263-
port2: undefined,
264-
promise: undefined,
265-
},
266-
};
253+
this[kState] = createReadableStreamState();
267254

268255
this[kIsClosedPromise] = createDeferredPromise();
269256
this[kControllerErrorFunction] = () => {};
@@ -655,17 +642,7 @@ function TransferredReadableStream() {
655642
return makeTransferable(ReflectConstruct(
656643
function() {
657644
this[kType] = 'ReadableStream';
658-
this[kState] = {
659-
disturbed: false,
660-
state: 'readable',
661-
storedError: undefined,
662-
stream: undefined,
663-
transfer: {
664-
writable: undefined,
665-
port: undefined,
666-
promise: undefined,
667-
},
668-
};
645+
this[kState] = createReadableStreamState();
669646
this[kIsClosedPromise] = createDeferredPromise();
670647
},
671648
[], ReadableStream));
@@ -1223,43 +1200,58 @@ ObjectDefineProperties(ReadableByteStreamController.prototype, {
12231200
[SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableByteStreamController.name),
12241201
});
12251202

1226-
function TeeReadableStream(start, pull, cancel) {
1203+
function InternalReadableStream(start, pull, cancel, highWaterMark, size) {
12271204
this[kType] = 'ReadableStream';
1228-
this[kState] = {
1229-
disturbed: false,
1230-
state: 'readable',
1231-
storedError: undefined,
1232-
stream: undefined,
1233-
transfer: {
1234-
writable: undefined,
1235-
port: undefined,
1236-
promise: undefined,
1237-
},
1238-
};
1205+
this[kState] = createReadableStreamState();
12391206
this[kIsClosedPromise] = createDeferredPromise();
1240-
setupReadableStreamDefaultControllerFromSource(
1207+
const controller = new ReadableStreamDefaultController(kSkipThrow);
1208+
setupReadableStreamDefaultController(
12411209
this,
1242-
ObjectCreate(null, {
1243-
start: { __proto__: null, value: start },
1244-
pull: { __proto__: null, value: pull },
1245-
cancel: { __proto__: null, value: cancel },
1246-
}),
1247-
1,
1248-
() => 1);
1210+
controller,
1211+
start,
1212+
pull,
1213+
cancel,
1214+
highWaterMark,
1215+
size);
1216+
return makeTransferable(this);
1217+
}
12491218

1219+
ObjectSetPrototypeOf(InternalReadableStream.prototype, ReadableStream.prototype);
1220+
ObjectSetPrototypeOf(InternalReadableStream, ReadableStream);
1221+
1222+
function createReadableStream(start, pull, cancel, highWaterMark = 1, size = () => 1) {
1223+
const stream = new InternalReadableStream(start, pull, cancel, highWaterMark, size);
1224+
1225+
// For spec compliance the InternalReadableStream must be a ReadableStream
1226+
stream.constructor = ReadableStream;
1227+
return stream;
1228+
}
12501229

1230+
function InternalReadableByteStream(start, pull, cancel) {
1231+
this[kType] = 'ReadableStream';
1232+
this[kState] = createReadableStreamState();
1233+
this[kIsClosedPromise] = createDeferredPromise();
1234+
const controller = new ReadableByteStreamController(kSkipThrow);
1235+
setupReadableByteStreamController(
1236+
this,
1237+
controller,
1238+
start,
1239+
pull,
1240+
cancel,
1241+
0,
1242+
undefined);
12511243
return makeTransferable(this);
12521244
}
12531245

1254-
ObjectSetPrototypeOf(TeeReadableStream.prototype, ReadableStream.prototype);
1255-
ObjectSetPrototypeOf(TeeReadableStream, ReadableStream);
1246+
ObjectSetPrototypeOf(InternalReadableByteStream.prototype, ReadableStream.prototype);
1247+
ObjectSetPrototypeOf(InternalReadableByteStream, ReadableStream);
12561248

1257-
function createTeeReadableStream(start, pull, cancel) {
1258-
const tee = new TeeReadableStream(start, pull, cancel);
1249+
function createReadableByteStream(start, pull, cancel) {
1250+
const stream = new InternalReadableByteStream(start, pull, cancel);
12591251

1260-
// For spec compliance the Tee must be a ReadableStream
1261-
tee.constructor = ReadableStream;
1262-
return tee;
1252+
// For spec compliance the InternalReadableByteStream must be a ReadableStream
1253+
stream.constructor = ReadableStream;
1254+
return stream;
12631255
}
12641256

12651257
const isReadableStream =
@@ -1275,6 +1267,23 @@ const isReadableStreamBYOBReader =
12751267

12761268
// ---- ReadableStream Implementation
12771269

1270+
function createReadableStreamState() {
1271+
return {
1272+
__proto__: null,
1273+
disturbed: false,
1274+
reader: undefined,
1275+
state: 'readable',
1276+
storedError: undefined,
1277+
transfer: {
1278+
__proto__: null,
1279+
writable: undefined,
1280+
port1: undefined,
1281+
port2: undefined,
1282+
promise: undefined,
1283+
},
1284+
};
1285+
}
1286+
12781287
function readableStreamFromIterable(iterable) {
12791288
let stream;
12801289
const iteratorRecord = getIterator(iterable, 'async');
@@ -1314,16 +1323,12 @@ function readableStreamFromIterable(iterable) {
13141323
});
13151324
}
13161325

1317-
stream = new ReadableStream({
1318-
start: startAlgorithm,
1319-
pull: pullAlgorithm,
1320-
cancel: cancelAlgorithm,
1321-
}, {
1322-
size() {
1323-
return 1;
1324-
},
1325-
highWaterMark: 0,
1326-
});
1326+
stream = createReadableStream(
1327+
startAlgorithm,
1328+
pullAlgorithm,
1329+
cancelAlgorithm,
1330+
0,
1331+
);
13271332

13281333
return stream;
13291334
}
@@ -1649,9 +1654,9 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
16491654
}
16501655

16511656
branch1 =
1652-
createTeeReadableStream(nonOpStart, pullAlgorithm, cancel1Algorithm);
1657+
createReadableStream(nonOpStart, pullAlgorithm, cancel1Algorithm);
16531658
branch2 =
1654-
createTeeReadableStream(nonOpStart, pullAlgorithm, cancel2Algorithm);
1659+
createReadableStream(nonOpStart, pullAlgorithm, cancel2Algorithm);
16551660

16561661
PromisePrototypeThen(
16571662
reader[kState].close.promise,
@@ -1928,16 +1933,10 @@ function readableByteStreamTee(stream) {
19281933
return cancelDeferred.promise;
19291934
}
19301935

1931-
branch1 = new ReadableStream({
1932-
type: 'bytes',
1933-
pull: pull1Algorithm,
1934-
cancel: cancel1Algorithm,
1935-
});
1936-
branch2 = new ReadableStream({
1937-
type: 'bytes',
1938-
pull: pull2Algorithm,
1939-
cancel: cancel2Algorithm,
1940-
});
1936+
branch1 =
1937+
createReadableByteStream(nonOpStart, pull1Algorithm, cancel1Algorithm);
1938+
branch2 =
1939+
createReadableByteStream(nonOpStart, pull2Algorithm, cancel2Algorithm);
19411940

19421941
forwardReaderError(reader);
19431942

@@ -1988,10 +1987,7 @@ function readableStreamCancel(stream, reason) {
19881987
}
19891988

19901989
return PromisePrototypeThen(
1991-
ensureIsPromise(
1992-
stream[kState].controller[kCancel],
1993-
stream[kState].controller,
1994-
reason),
1990+
stream[kState].controller[kCancel](reason),
19951991
() => {});
19961992
}
19971993

@@ -2356,7 +2352,7 @@ function readableStreamDefaultControllerCallPullIfNeeded(controller) {
23562352
assert(!controller[kState].pullAgain);
23572353
controller[kState].pulling = true;
23582354
PromisePrototypeThen(
2359-
ensureIsPromise(controller[kState].pullAlgorithm, controller),
2355+
controller[kState].pullAlgorithm(controller),
23602356
() => {
23612357
controller[kState].pulling = false;
23622358
if (controller[kState].pullAgain) {
@@ -2386,12 +2382,9 @@ function readableStreamDefaultControllerError(controller, error) {
23862382

23872383
function readableStreamDefaultControllerCancelSteps(controller, reason) {
23882384
resetQueue(controller);
2389-
try {
2390-
const result = controller[kState].cancelAlgorithm(reason);
2391-
return result;
2392-
} finally {
2393-
readableStreamDefaultControllerClearAlgorithms(controller);
2394-
}
2385+
const result = controller[kState].cancelAlgorithm(reason);
2386+
readableStreamDefaultControllerClearAlgorithms(controller);
2387+
return result;
23952388
}
23962389

23972390
function readableStreamDefaultControllerPullSteps(controller, readRequest) {
@@ -2465,11 +2458,10 @@ function setupReadableStreamDefaultControllerFromSource(
24652458
FunctionPrototypeBind(start, source, controller) :
24662459
nonOpStart;
24672460
const pullAlgorithm = pull ?
2468-
FunctionPrototypeBind(pull, source, controller) :
2461+
createPromiseCallback('source.pull', pull, source) :
24692462
nonOpPull;
2470-
24712463
const cancelAlgorithm = cancel ?
2472-
FunctionPrototypeBind(cancel, source) :
2464+
createPromiseCallback('source.cancel', cancel, source) :
24732465
nonOpCancel;
24742466

24752467
setupReadableStreamDefaultController(
@@ -3097,7 +3089,7 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
30973089
assert(!controller[kState].pullAgain);
30983090
controller[kState].pulling = true;
30993091
PromisePrototypeThen(
3100-
ensureIsPromise(controller[kState].pullAlgorithm, controller),
3092+
controller[kState].pullAlgorithm(controller),
31013093
() => {
31023094
controller[kState].pulling = false;
31033095
if (controller[kState].pullAgain) {
@@ -3264,10 +3256,10 @@ function setupReadableByteStreamControllerFromSource(
32643256
FunctionPrototypeBind(start, source, controller) :
32653257
nonOpStart;
32663258
const pullAlgorithm = pull ?
3267-
FunctionPrototypeBind(pull, source, controller) :
3259+
createPromiseCallback('source.pull', pull, source, controller) :
32683260
nonOpPull;
32693261
const cancelAlgorithm = cancel ?
3270-
FunctionPrototypeBind(cancel, source) :
3262+
createPromiseCallback('source.cancel', cancel, source) :
32713263
nonOpCancel;
32723264

32733265
if (autoAllocateChunkSize === 0) {
@@ -3364,4 +3356,6 @@ module.exports = {
33643356
readableByteStreamControllerPullSteps,
33653357
setupReadableByteStreamController,
33663358
setupReadableByteStreamControllerFromSource,
3359+
createReadableStream,
3360+
createReadableByteStream,
33673361
};

0 commit comments

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