OLD | NEW |
(Empty) | |
| 1 (function(global, exports) { |
| 2 'use strict'; |
| 3 |
| 4 // V8 "Imports": |
| 5 // - %CreatePrivateSymbol |
| 6 // - %_CallFunction |
| 7 // - %AddNamedProperty |
| 8 // - %HasOwnProperty |
| 9 // - $promiseThen, $promiseCreate, $promiseResolve, $promiseReject |
| 10 |
| 11 const readableStreamCloseRequested = %CreatePrivateSymbol('[[closeRequested]
]'); |
| 12 const readableStreamController = %CreatePrivateSymbol('[[controller]]'); |
| 13 const readableStreamPullAgain = %CreatePrivateSymbol('[[pullAgain]]'); |
| 14 const readableStreamPulling = %CreatePrivateSymbol('[[pulling]]'); |
| 15 const readableStreamQueue = %CreatePrivateSymbol('[[queue]]'); |
| 16 const readableStreamQueueSize = %CreatePrivateSymbol('[[queue]] total size')
; |
| 17 const readableStreamReader = %CreatePrivateSymbol('[[reader]]'); |
| 18 const readableStreamStarted = %CreatePrivateSymbol('[[started]]'); |
| 19 const readableStreamState = %CreatePrivateSymbol('[[state]]'); |
| 20 const readableStreamStoredError = %CreatePrivateSymbol('[[storedError]]'); |
| 21 const readableStreamStrategySize = %CreatePrivateSymbol('[[strategySize]]'); |
| 22 const readableStreamStrategyHWM = %CreatePrivateSymbol('[[strategyHWM]]'); |
| 23 const readableStreamUnderlyingSource = %CreatePrivateSymbol('[[underlyingSou
rce]]'); |
| 24 |
| 25 const readableStreamControllerControlledReadableStream = %CreatePrivateSymbo
l('[[controlledReadableStream]]'); |
| 26 |
| 27 const readableStreamReaderClosedPromise = %CreatePrivateSymbol('[[closedProm
ise]]'); |
| 28 const readableStreamReaderOwnerReadableStream = %CreatePrivateSymbol('[[owne
rReadableStream]]'); |
| 29 const readableStreamReaderReadRequests = %CreatePrivateSymbol('[[readRequest
s]]'); |
| 30 const readableStreamReaderState = %CreatePrivateSymbol('[[state]]'); |
| 31 const readableStreamReaderStoredError = %CreatePrivateSymbol('[[storedError]
]'); |
| 32 |
| 33 const createWithExternalControllerSentinel = %CreatePrivateSymbol( |
| 34 'flag for UA-controlled ReadableStreams to pass'); |
| 35 |
| 36 const undefined = void 0; |
| 37 const DONT_ENUM = 2; |
| 38 |
| 39 const STATE_READABLE = 0; |
| 40 const STATE_CLOSED = 1; |
| 41 const STATE_ERRORED = 2; |
| 42 |
| 43 const TypeError = global.TypeError; |
| 44 const RangeError = global.RangeError; |
| 45 const Promise = global.Promise; |
| 46 |
| 47 const Number = global.Number; |
| 48 const Number_isNaN = global.Number.isNaN; |
| 49 const Number_isFinite = global.Number.isFinite; |
| 50 |
| 51 // TODO(domenic): update to use InternalPackedArray once yangguo gives us ac
cess in extras |
| 52 const InternalPackedArray = global.Array; |
| 53 |
| 54 function thenPromise(promise, f, r) { |
| 55 return %_CallFunction(promise, f, r, $promiseThen); |
| 56 } |
| 57 |
| 58 // Manually create "bound" versions since Function.prototype.bind is slow. |
| 59 const Promise_resolve = (function() { |
| 60 const unbound = Promise.resolve; |
| 61 return function(x) { |
| 62 return %_CallFunction(Promise, x, unbound); |
| 63 }; |
| 64 })(); |
| 65 |
| 66 const Promise_reject = (function() { |
| 67 const unbound = Promise.reject; |
| 68 return function(r) { |
| 69 return %_CallFunction(Promise, r, unbound); |
| 70 }; |
| 71 })(); |
| 72 |
| 73 class ReadableStream { |
| 74 constructor(underlyingSource, strategy) { |
| 75 if (underlyingSource === undefined) { |
| 76 underlyingSource = {}; |
| 77 } |
| 78 if (strategy === undefined) { |
| 79 strategy = {}; |
| 80 } |
| 81 const size = strategy.size; |
| 82 let highWaterMark = strategy.highWaterMark; |
| 83 if (highWaterMark === undefined) { |
| 84 highWaterMark = 1; |
| 85 } |
| 86 |
| 87 const normalizedStrategy = ValidateAndNormalizeQueuingStrategy(size,
highWaterMark); |
| 88 |
| 89 this[readableStreamUnderlyingSource] = underlyingSource; |
| 90 |
| 91 this[readableStreamQueue] = new InternalPackedArray(); |
| 92 this[readableStreamQueueSize] = 0; |
| 93 |
| 94 // TODO(domenic) consolidate booleans into a bit field? |
| 95 // TODO(domenic) use integers for state? (or put in bit field?) |
| 96 this[readableStreamState] = STATE_READABLE; |
| 97 this[readableStreamStarted] = false; |
| 98 this[readableStreamCloseRequested] = false; |
| 99 this[readableStreamPulling] = false; |
| 100 this[readableStreamPullAgain] = false; |
| 101 this[readableStreamReader] = undefined; |
| 102 |
| 103 this[readableStreamStoredError] = undefined; |
| 104 this[readableStreamStrategySize] = normalizedStrategy.size; |
| 105 this[readableStreamStrategyHWM] = normalizedStrategy.highWaterMark; |
| 106 |
| 107 // Avoid allocating a controller if the stream is going to be contro
lled externally (i.e. from C++) anyway. |
| 108 // All calls to underlyingSource methods will disregard their contro
ller argument in such situations |
| 109 // (but see below). |
| 110 const controller = arguments[2] === createWithExternalControllerSent
inel ? |
| 111 null : |
| 112 new ReadableStreamController(this); |
| 113 this[readableStreamController] = controller; |
| 114 |
| 115 // We need to pass ourself to the underlyingSource start method for
externally-controlled streams. We |
| 116 // use the now-useless controller argument to do so. |
| 117 const argToStart = arguments[2] === createWithExternalControllerSent
inel ? this : controller; |
| 118 |
| 119 const that = this; |
| 120 const startResult = CallOrNoop(underlyingSource, 'start', argToStart
, 'underlyingSource.start'); |
| 121 thenPromise(Promise_resolve(startResult), |
| 122 function() { |
| 123 that[readableStreamStarted] = true; |
| 124 RequestReadableStreamPull(that); |
| 125 }, |
| 126 function(r) { |
| 127 if (that[readableStreamState] === STATE_READABLE) { |
| 128 return ErrorReadableStream(that, r); |
| 129 } |
| 130 } |
| 131 ); |
| 132 } |
| 133 |
| 134 cancel(reason) { |
| 135 if (IsReadableStream(this) === false) { |
| 136 return Promise_reject(new TypeError( |
| 137 'ReadableStream.prototype.cancel can only be used on a Reada
bleStream')); |
| 138 } |
| 139 |
| 140 if (IsReadableStreamLocked(this) === true) { |
| 141 return Promise_reject(new TypeError( |
| 142 'Cannot cancel a stream that already has a reader')); |
| 143 } |
| 144 |
| 145 return CancelReadableStream(this, reason); |
| 146 } |
| 147 |
| 148 getReader() { |
| 149 if (IsReadableStream(this) === false) { |
| 150 throw new TypeError('ReadableStream.prototype.getReader can only
be used on a ReadableStream'); |
| 151 } |
| 152 |
| 153 return AcquireReadableStreamReader(this); |
| 154 } |
| 155 |
| 156 tee() { |
| 157 if (IsReadableStream(this) === false) { |
| 158 throw new TypeError('ReadableStream.prototype.tee can only be us
ed on a ReadableStream'); |
| 159 } |
| 160 |
| 161 return TeeReadableStream(this); |
| 162 } |
| 163 } |
| 164 |
| 165 class ReadableStreamController { |
| 166 constructor(stream) { |
| 167 if (IsReadableStream(stream) === false) { |
| 168 throw new TypeError('ReadableStreamController can only be constr
ucted with a ReadableStream instance'); |
| 169 } |
| 170 |
| 171 if (stream[readableStreamController] !== undefined) { |
| 172 throw new TypeError( |
| 173 'ReadableStreamController instances can only be created by t
he ReadableStream constructor'); |
| 174 } |
| 175 |
| 176 this[readableStreamControllerControlledReadableStream] = stream; |
| 177 } |
| 178 |
| 179 get desiredSize() { |
| 180 if (IsReadableStreamController(this) === false) { |
| 181 throw new TypeError( |
| 182 'ReadableStreamController.prototype.desiredSize can only be
used on a ReadableStreamController'); |
| 183 } |
| 184 |
| 185 return GetReadableStreamDesiredSize(this[readableStreamControllerCon
trolledReadableStream]); |
| 186 } |
| 187 |
| 188 close() { |
| 189 if (IsReadableStreamController(this) === false) { |
| 190 throw new TypeError( |
| 191 'ReadableStreamController.prototype.close can only be used o
n a ReadableStreamController'); |
| 192 } |
| 193 |
| 194 const stream = this[readableStreamControllerControlledReadableStream
]; |
| 195 |
| 196 if (stream[readableStreamCloseRequested] === true) { |
| 197 throw new TypeError('The stream has already been closed; do not
close it again!'); |
| 198 } |
| 199 if (stream[readableStreamState] === STATE_ERRORED) { |
| 200 throw new TypeError('The stream is in an errored state and canno
t be closed'); |
| 201 } |
| 202 |
| 203 return CloseReadableStream(stream); |
| 204 } |
| 205 |
| 206 enqueue(chunk) { |
| 207 if (IsReadableStreamController(this) === false) { |
| 208 throw new TypeError( |
| 209 'ReadableStreamController.prototype.enqueue can only be used
on a ReadableStreamController'); |
| 210 } |
| 211 |
| 212 const stream = this[readableStreamControllerControlledReadableStream
]; |
| 213 |
| 214 if (stream[readableStreamState] === STATE_ERRORED) { |
| 215 throw stream[readableStreamStoredError]; |
| 216 } |
| 217 |
| 218 if (stream[readableStreamCloseRequested] === true) { |
| 219 throw new TypeError('stream is closed or draining'); |
| 220 } |
| 221 |
| 222 return EnqueueInReadableStream(stream, chunk); |
| 223 } |
| 224 |
| 225 error(e) { |
| 226 if (IsReadableStreamController(this) === false) { |
| 227 throw new TypeError( |
| 228 'ReadableStreamController.prototype.error can only be used o
n a ReadableStreamController'); |
| 229 } |
| 230 |
| 231 const stream = this[readableStreamControllerControlledReadableStream
]; |
| 232 |
| 233 const state = stream[readableStreamState]; |
| 234 if (state !== STATE_READABLE) { |
| 235 throw new TypeError(`The stream is ${state} and so cannot be err
ored`); |
| 236 } |
| 237 |
| 238 return ErrorReadableStream(stream, e); |
| 239 } |
| 240 } |
| 241 |
| 242 class ReadableStreamReader { |
| 243 constructor(stream) { |
| 244 if (IsReadableStream(stream) === false) { |
| 245 throw new TypeError('ReadableStreamReader can only be constructed
with a ReadableStream instance'); |
| 246 } |
| 247 if (IsReadableStreamLocked(stream) === true) { |
| 248 throw new TypeError('This stream has already been locked for exclu
sive reading by another reader'); |
| 249 } |
| 250 |
| 251 stream[readableStreamReader] = this; |
| 252 this[readableStreamReaderOwnerReadableStream] = stream; |
| 253 |
| 254 // TODO(domenic): use integers for state? |
| 255 this[readableStreamReaderState] = STATE_READABLE; |
| 256 this[readableStreamReaderStoredError] = undefined; |
| 257 this[readableStreamReaderReadRequests] = new InternalPackedArray(); |
| 258 this[readableStreamReaderClosedPromise] = $promiseCreate(); |
| 259 |
| 260 const streamState = stream[readableStreamState]; |
| 261 if (streamState === STATE_CLOSED || streamState === STATE_ERRORED) { |
| 262 ReleaseReadableStreamReader(this); |
| 263 } |
| 264 } |
| 265 |
| 266 get closed() { |
| 267 if (IsReadableStreamReader(this) === false) { |
| 268 return Promise_reject( |
| 269 new TypeError('ReadableStreamReader.prototype.closed can onl
y be used on a ReadableStreamReader')); |
| 270 } |
| 271 |
| 272 return this[readableStreamReaderClosedPromise]; |
| 273 } |
| 274 |
| 275 cancel(reason) { |
| 276 if (IsReadableStreamReader(this) === false) { |
| 277 return Promise_reject( |
| 278 new TypeError('ReadableStreamReader.prototype.cancel can onl
y be used on a ReadableStreamReader')); |
| 279 } |
| 280 |
| 281 const state = this[readableStreamReaderState]; |
| 282 if (state === STATE_CLOSED) { |
| 283 return Promise_resolve(undefined); |
| 284 } |
| 285 |
| 286 if (state === STATE_ERRORED) { |
| 287 return Promise_reject(this[readableStreamReaderStoredError]); |
| 288 } |
| 289 |
| 290 return CancelReadableStream(this[readableStreamReaderOwnerReadableSt
ream], reason); |
| 291 } |
| 292 |
| 293 read() { |
| 294 if (IsReadableStreamReader(this) === false) { |
| 295 return Promise_reject( |
| 296 new TypeError('ReadableStreamReader.prototype.read can only
be used on a ReadableStreamReader')); |
| 297 } |
| 298 |
| 299 return ReadFromReadableStreamReader(this); |
| 300 } |
| 301 |
| 302 releaseLock() { |
| 303 if (IsReadableStreamReader(this) === false) { |
| 304 throw new TypeError( |
| 305 'ReadableStreamReader.prototype.releaseLock can only be used
on a ReadableStreamReader'); |
| 306 } |
| 307 |
| 308 if (this[readableStreamReaderOwnerReadableStream] === undefined) { |
| 309 return undefined; |
| 310 } |
| 311 |
| 312 if (this[readableStreamReaderReadRequests].length > 0) { |
| 313 throw new TypeError( |
| 314 'Tried to release a reader lock when that reader has pending
read() calls un-settled'); |
| 315 } |
| 316 |
| 317 return ReleaseReadableStreamReader(this); |
| 318 } |
| 319 } |
| 320 |
| 321 // |
| 322 // Readable stream abstract operations |
| 323 // |
| 324 |
| 325 function AcquireReadableStreamReader(stream) { |
| 326 return new ReadableStreamReader(stream); |
| 327 } |
| 328 |
| 329 function CancelReadableStream(stream, reason) { |
| 330 const state = stream[readableStreamState]; |
| 331 if (state === STATE_CLOSED) { |
| 332 return Promise_resolve(undefined); |
| 333 } |
| 334 if (state === STATE_ERRORED) { |
| 335 return Promise_reject(stream[readableStreamStoredError]); |
| 336 } |
| 337 |
| 338 stream[readableStreamQueue] = new InternalPackedArray(); |
| 339 FinishClosingReadableStream(stream); |
| 340 |
| 341 const underlyingSource = stream[readableStreamUnderlyingSource]; |
| 342 const sourceCancelPromise = PromiseCallOrNoop(underlyingSource, 'cancel'
, reason, 'underlyingSource.cancel'); |
| 343 return thenPromise(sourceCancelPromise, function() { return undefined; }
); |
| 344 } |
| 345 |
| 346 function CloseReadableStream(stream) { |
| 347 if (stream[readableStreamState] === STATE_CLOSED) { |
| 348 return undefined; |
| 349 } |
| 350 |
| 351 stream[readableStreamCloseRequested] = true; |
| 352 |
| 353 if (stream[readableStreamQueue].length === 0) { |
| 354 return FinishClosingReadableStream(stream); |
| 355 } |
| 356 } |
| 357 |
| 358 |
| 359 function EnqueueInReadableStream(stream, chunk) { |
| 360 if (stream[readableStreamState] === STATE_CLOSED) { |
| 361 return undefined; |
| 362 } |
| 363 |
| 364 if (IsReadableStreamLocked(stream) === true && |
| 365 stream[readableStreamReader][readableStreamReaderReadRequests].lengt
h > 0) { |
| 366 const readRequest = stream[readableStreamReader][readableStreamReade
rReadRequests].shift(); |
| 367 $promiseResolve(readRequest, CreateIterResultObject(chunk, false)); |
| 368 } else { |
| 369 let chunkSize = 1; |
| 370 |
| 371 const strategySize = stream[readableStreamStrategySize]; |
| 372 if (strategySize !== undefined) { |
| 373 try { |
| 374 chunkSize = strategySize(chunk); |
| 375 } catch (chunkSizeE) { |
| 376 ErrorReadableStream(stream, chunkSizeE); |
| 377 throw chunkSizeE; |
| 378 } |
| 379 } |
| 380 |
| 381 try { |
| 382 EnqueueValueWithSize(stream, chunk, chunkSize); |
| 383 } catch (enqueueE) { |
| 384 ErrorReadableStream(stream, enqueueE); |
| 385 throw enqueueE; |
| 386 } |
| 387 } |
| 388 |
| 389 RequestReadableStreamPull(stream); |
| 390 } |
| 391 |
| 392 function ErrorReadableStream(stream, e) { |
| 393 stream[readableStreamQueue] = new InternalPackedArray(); |
| 394 stream[readableStreamStoredError] = e; |
| 395 stream[readableStreamState] = STATE_ERRORED; |
| 396 |
| 397 if (IsReadableStreamLocked(stream) === true) { |
| 398 return ReleaseReadableStreamReader(stream[readableStreamReader]); |
| 399 } |
| 400 } |
| 401 |
| 402 function FinishClosingReadableStream(stream) { |
| 403 stream[readableStreamState] = STATE_CLOSED; |
| 404 |
| 405 if (IsReadableStreamLocked(stream) === true) { |
| 406 return ReleaseReadableStreamReader(stream[readableStreamReader]); |
| 407 } |
| 408 } |
| 409 |
| 410 function GetReadableStreamDesiredSize(stream) { |
| 411 const queueSize = GetTotalQueueSize(stream); |
| 412 return stream[readableStreamStrategyHWM] - queueSize; |
| 413 } |
| 414 |
| 415 function IsReadableStream(x) { |
| 416 return %HasOwnProperty(x, readableStreamUnderlyingSource); |
| 417 } |
| 418 |
| 419 function IsReadableStreamLocked(stream) { |
| 420 return stream[readableStreamReader] !== undefined; |
| 421 } |
| 422 |
| 423 function IsReadableStreamController(x) { |
| 424 return %HasOwnProperty(x, readableStreamControllerControlledReadableStre
am); |
| 425 } |
| 426 |
| 427 function IsReadableStreamReader(x) { |
| 428 return %HasOwnProperty(x, readableStreamReaderOwnerReadableStream); |
| 429 } |
| 430 |
| 431 function ReadFromReadableStreamReader(reader) { |
| 432 const state = reader[readableStreamReaderState]; |
| 433 if (state === STATE_CLOSED) { |
| 434 return Promise_resolve(CreateIterResultObject(undefined, true)); |
| 435 } |
| 436 |
| 437 if (state === STATE_ERRORED) { |
| 438 return Promise_reject(reader[readableStreamReaderStoredError]); |
| 439 } |
| 440 |
| 441 const ownerReadableStream = reader[readableStreamReaderOwnerReadableStre
am]; |
| 442 const queue = ownerReadableStream[readableStreamQueue]; |
| 443 if (queue.length > 0) { |
| 444 const chunk = DequeueValue(ownerReadableStream); |
| 445 |
| 446 if (ownerReadableStream[readableStreamCloseRequested] === true && qu
eue.length === 0) { |
| 447 FinishClosingReadableStream(ownerReadableStream); |
| 448 } else { |
| 449 RequestReadableStreamPull(ownerReadableStream); |
| 450 } |
| 451 |
| 452 return Promise_resolve(CreateIterResultObject(chunk, false)); |
| 453 } else { |
| 454 const readRequest = $promiseCreate(); |
| 455 |
| 456 reader[readableStreamReaderReadRequests].push(readRequest); |
| 457 RequestReadableStreamPull(ownerReadableStream); |
| 458 return readRequest; |
| 459 } |
| 460 } |
| 461 |
| 462 function ReleaseReadableStreamReader(reader) { |
| 463 const ownerReadableStream = reader[readableStreamReaderOwnerReadableStre
am]; |
| 464 if (ownerReadableStream[readableStreamState] === STATE_ERRORED) { |
| 465 reader[readableStreamReaderState] = STATE_ERRORED; |
| 466 |
| 467 const e = ownerReadableStream[readableStreamStoredError]; |
| 468 reader[readableStreamReaderStoredError] = e; |
| 469 $promiseReject(reader[readableStreamReaderClosedPromise], e); |
| 470 |
| 471 const readRequests = reader[readableStreamReaderReadRequests]; |
| 472 for (let i = 0; i < readRequests.length; ++i) { |
| 473 $promiseReject(readRequests[i], e); |
| 474 } |
| 475 } else { |
| 476 reader[readableStreamReaderState] = STATE_CLOSED; |
| 477 $promiseResolve(reader[readableStreamReaderClosedPromise], undefined
); |
| 478 |
| 479 const readRequests = reader[readableStreamReaderReadRequests]; |
| 480 for (let i = 0; i < readRequests.length; ++i) { |
| 481 $promiseResolve(readRequests[i], CreateIterResultObject(undefine
d, true)); |
| 482 } |
| 483 } |
| 484 |
| 485 reader[readableStreamReaderReadRequests] = new InternalPackedArray(); |
| 486 ownerReadableStream[readableStreamReader] = undefined; |
| 487 reader[readableStreamReaderOwnerReadableStream] = undefined; |
| 488 } |
| 489 |
| 490 function RequestReadableStreamPull(stream) { |
| 491 const shouldPull = ShouldReadableStreamPull(stream); |
| 492 if (shouldPull === false) { |
| 493 return undefined; |
| 494 } |
| 495 |
| 496 if (stream[readableStreamPulling] === true) { |
| 497 stream[readableStreamPullAgain] = true; |
| 498 return undefined; |
| 499 } |
| 500 |
| 501 stream[readableStreamPulling] = true; |
| 502 |
| 503 const underlyingSource = stream[readableStreamUnderlyingSource]; |
| 504 const controller = stream[readableStreamController]; |
| 505 const pullPromise = PromiseCallOrNoop(underlyingSource, 'pull', controll
er, 'underlyingSource.pull'); |
| 506 |
| 507 thenPromise(pullPromise, |
| 508 function() { |
| 509 stream[readableStreamPulling] = false; |
| 510 |
| 511 if (stream[readableStreamPullAgain] === true) { |
| 512 stream[readableStreamPullAgain] = false; |
| 513 return RequestReadableStreamPull(stream); |
| 514 } |
| 515 }, |
| 516 function(e) { |
| 517 if (stream[readableStreamState] === STATE_READABLE) { |
| 518 return ErrorReadableStream(stream, e); |
| 519 } |
| 520 } |
| 521 ); |
| 522 } |
| 523 |
| 524 function ShouldReadableStreamPull(stream) { |
| 525 const state = stream[readableStreamState]; |
| 526 if (state === STATE_CLOSED || state === STATE_ERRORED) { |
| 527 return false; |
| 528 } |
| 529 |
| 530 if (stream[readableStreamCloseRequested] === true) { |
| 531 return false; |
| 532 } |
| 533 |
| 534 if (stream[readableStreamStarted] === false) { |
| 535 return false; |
| 536 } |
| 537 |
| 538 if (IsReadableStreamLocked(stream) === true) { |
| 539 const reader = stream[readableStreamReader]; |
| 540 const readRequests = reader[readableStreamReaderReadRequests]; |
| 541 if (readRequests.length > 0) { |
| 542 return true; |
| 543 } |
| 544 } |
| 545 |
| 546 const desiredSize = GetReadableStreamDesiredSize(stream); |
| 547 if (desiredSize > 0) { |
| 548 return true; |
| 549 } |
| 550 |
| 551 return false; |
| 552 } |
| 553 |
| 554 // Potential future optimization: use class instances for the underlying sou
rces, so that we don't re-create |
| 555 // closures every time. |
| 556 |
| 557 function TeeReadableStream(stream) { // shouldClone argument from spec not s
upported yet |
| 558 const reader = AcquireReadableStreamReader(stream); |
| 559 |
| 560 let closedOrErrored = false; |
| 561 let canceled1 = false; |
| 562 let canceled2 = false; |
| 563 let reason1; |
| 564 let reason2; |
| 565 let promise = $promiseCreate(); |
| 566 |
| 567 const branch1 = new ReadableStream({ |
| 568 pull, |
| 569 cancel: cancel1 |
| 570 }); |
| 571 |
| 572 const branch2 = new ReadableStream({ |
| 573 pull, |
| 574 cancel: cancel2 |
| 575 }); |
| 576 |
| 577 thenPromise(reader[readableStreamReaderClosedPromise], undefined, functi
on (r) { |
| 578 if (closedOrErrored === true) { |
| 579 return; |
| 580 } |
| 581 |
| 582 ErrorReadableStream(branch1, r); |
| 583 ErrorReadableStream(branch2, r); |
| 584 closedOrErrored = true; |
| 585 }); |
| 586 |
| 587 return [branch1, branch2]; |
| 588 |
| 589 |
| 590 function pull() { |
| 591 return thenPromise(ReadFromReadableStreamReader(reader), function (r
esult) { |
| 592 const value = result.value; |
| 593 const done = result.done; |
| 594 |
| 595 if (done === true && closedOrErrored === false) { |
| 596 CloseReadableStream(branch1); |
| 597 CloseReadableStream(branch2); |
| 598 closedOrErrored = true; |
| 599 } |
| 600 |
| 601 if (closedOrErrored === true) { |
| 602 return; |
| 603 } |
| 604 |
| 605 if (canceled1 === false) { |
| 606 EnqueueInReadableStream(branch1, value); |
| 607 } |
| 608 |
| 609 if (canceled2 === false) { |
| 610 EnqueueInReadableStream(branch2, value); |
| 611 } |
| 612 }); |
| 613 } |
| 614 |
| 615 function cancel1(reason) { |
| 616 canceled1 = true; |
| 617 reason1 = reason; |
| 618 |
| 619 if (canceled2 === true) { |
| 620 const compositeReason = [reason1, reason2]; |
| 621 const cancelResult = CancelReadableStream(stream, compositeReaso
n); |
| 622 $promiseResolve(promise, cancelResult); |
| 623 } |
| 624 |
| 625 return promise; |
| 626 } |
| 627 |
| 628 function cancel2(reason) { |
| 629 canceled2 = true; |
| 630 reason2 = reason; |
| 631 |
| 632 if (canceled1 === true) { |
| 633 const compositeReason = [reason1, reason2]; |
| 634 const cancelResult = CancelReadableStream(stream, compositeReaso
n); |
| 635 $promiseResolve(promise, cancelResult); |
| 636 } |
| 637 |
| 638 return promise; |
| 639 } |
| 640 } |
| 641 |
| 642 // |
| 643 // Queue-with-sizes |
| 644 // Modified from taking the queue (as in the spec) to taking the stream, so
we can modify the queue size alongside. |
| 645 // |
| 646 |
| 647 function DequeueValue(stream) { |
| 648 const result = stream[readableStreamQueue].shift(); |
| 649 stream[readableStreamQueueSize] -= result.size; |
| 650 return result.value; |
| 651 } |
| 652 |
| 653 function EnqueueValueWithSize(stream, value, size) { |
| 654 size = Number(size); |
| 655 if (!Number_isFinite(size)) { |
| 656 throw new RangeError('size must be a finite, non-NaN number.'); |
| 657 } |
| 658 |
| 659 stream[readableStreamQueueSize] += size; |
| 660 stream[readableStreamQueue].push({ value, size }); |
| 661 } |
| 662 |
| 663 function GetTotalQueueSize(stream) { |
| 664 return stream[readableStreamQueueSize]; |
| 665 } |
| 666 |
| 667 // |
| 668 // Other helpers |
| 669 // |
| 670 |
| 671 function ValidateAndNormalizeQueuingStrategy(size, highWaterMark) { |
| 672 if (size !== undefined && typeof size !== 'function') { |
| 673 throw new TypeError('size property of a queuing strategy must be a f
unction'); |
| 674 } |
| 675 |
| 676 highWaterMark = Number(highWaterMark); |
| 677 if (Number_isNaN(highWaterMark)) { |
| 678 throw new TypeError('highWaterMark property of a queuing strategy mu
st be convertible to a non-NaN number'); |
| 679 } |
| 680 if (highWaterMark < 0) { |
| 681 throw new RangeError('highWaterMark property of a queuing strategy m
ust be nonnegative'); |
| 682 } |
| 683 |
| 684 return { size, highWaterMark }; |
| 685 } |
| 686 |
| 687 function CallOrNoop(O, P, arg, nameForError) { // Modified from InvokeOrNoo
p in spec |
| 688 const method = O[P]; |
| 689 if (method === undefined) { |
| 690 return undefined; |
| 691 } |
| 692 if (typeof method !== 'function') { |
| 693 throw new TypeError(`${nameForError} must be a function or undefined
`); |
| 694 } |
| 695 |
| 696 return %_CallFunction(O, arg, method); |
| 697 } |
| 698 |
| 699 |
| 700 function PromiseCallOrNoop(O, P, arg, nameForError) { // Modified from Promi
seInvokeOrNoop in spec |
| 701 let method; |
| 702 try { |
| 703 method = O[P]; |
| 704 } catch (methodE) { |
| 705 return Promise_reject(methodE); |
| 706 } |
| 707 |
| 708 if (method === undefined) { |
| 709 return Promise_resolve(undefined); |
| 710 } |
| 711 |
| 712 if (typeof method !== 'function') { |
| 713 return Promise_reject(`${nameForError} must be a function or undefin
ed`); |
| 714 } |
| 715 |
| 716 try { |
| 717 return Promise_resolve(%_CallFunction(O, arg, method)); |
| 718 } catch (e) { |
| 719 return Promise_reject(e); |
| 720 } |
| 721 } |
| 722 |
| 723 function CreateIterResultObject(value, done) { |
| 724 return { value, done }; |
| 725 } |
| 726 |
| 727 |
| 728 // |
| 729 // Additions to the global |
| 730 // |
| 731 |
| 732 %AddNamedProperty(global, 'ReadableStream', ReadableStream, DONT_ENUM); |
| 733 |
| 734 // |
| 735 // Exports for Blink to use |
| 736 // |
| 737 |
| 738 exports.ReadableStream = ReadableStream; |
| 739 exports.createWithExternalControllerSentinel = createWithExternalControllerS
entinel; |
| 740 exports.ErrorReadableStream = ErrorReadableStream; |
| 741 exports.EnqueueInReadableStream = EnqueueInReadableStream; |
| 742 exports.CloseReadableStream = CloseReadableStream; |
| 743 exports.GetReadableStreamDesiredSize = GetReadableStreamDesiredSize; |
| 744 }); |
OLD | NEW |