OLD | NEW |
(Empty) | |
| 1 // Copyright 2015 The Chromium Authors. All rights reserved. |
| 2 // Use of this source code is governed by a BSD-style license that can be |
| 3 // found in the LICENSE file. |
| 4 |
| 5 (function(global, binding, v8) { |
| 6 'use strict'; |
| 7 |
| 8 const readableStreamController = v8.createPrivateSymbol('[[controller]]'); |
| 9 const readableStreamQueue = v8.createPrivateSymbol('[[queue]]'); |
| 10 const readableStreamQueueSize = |
| 11 v8.createPrivateSymbol('[[queue]] total size'); |
| 12 const readableStreamReader = v8.createPrivateSymbol('[[reader]]'); |
| 13 const readableStreamState = v8.createPrivateSymbol('[[state]]'); |
| 14 const readableStreamStoredError = v8.createPrivateSymbol('[[storedError]]'); |
| 15 const readableStreamStrategySize = v8.createPrivateSymbol('[[strategySize]]'); |
| 16 const readableStreamStrategyHWM = v8.createPrivateSymbol('[[strategyHWM]]'); |
| 17 const readableStreamUnderlyingSource = |
| 18 v8.createPrivateSymbol('[[underlyingSource]]'); |
| 19 |
| 20 const readableStreamControllerControlledReadableStream = |
| 21 v8.createPrivateSymbol('[[controlledReadableStream]]'); |
| 22 |
| 23 const readableStreamReaderClosedPromise = |
| 24 v8.createPrivateSymbol('[[closedPromise]]'); |
| 25 const readableStreamReaderOwnerReadableStream = |
| 26 v8.createPrivateSymbol('[[ownerReadableStream]]'); |
| 27 const readableStreamReaderReadRequests = |
| 28 v8.createPrivateSymbol('[[readRequests]]'); |
| 29 |
| 30 const STATE_READABLE = 0; |
| 31 const STATE_CLOSED = 1; |
| 32 const STATE_ERRORED = 2; |
| 33 |
| 34 const readableStreamBits = v8.createPrivateSymbol( |
| 35 'bit field for [[started]], [[closeRequested]], [[pulling]], [[pullAgain]]
, [[disturbed]]'); |
| 36 const STARTED = 0b1; |
| 37 const CLOSE_REQUESTED = 0b10; |
| 38 const PULLING = 0b100; |
| 39 const PULL_AGAIN = 0b1000; |
| 40 const DISTURBED = 0b10000; |
| 41 |
| 42 const undefined = global.undefined; |
| 43 const Infinity = global.Infinity; |
| 44 |
| 45 const defineProperty = global.Object.defineProperty; |
| 46 const hasOwnProperty = v8.uncurryThis(global.Object.hasOwnProperty); |
| 47 const callFunction = v8.uncurryThis(global.Function.prototype.call); |
| 48 |
| 49 const TypeError = global.TypeError; |
| 50 const RangeError = global.RangeError; |
| 51 |
| 52 const Number = global.Number; |
| 53 const Number_isNaN = Number.isNaN; |
| 54 const Number_isFinite = Number.isFinite; |
| 55 |
| 56 const Promise = global.Promise; |
| 57 const thenPromise = v8.uncurryThis(Promise.prototype.then); |
| 58 const Promise_resolve = v8.simpleBind(Promise.resolve, Promise); |
| 59 const Promise_reject = v8.simpleBind(Promise.reject, Promise); |
| 60 |
| 61 const errIllegalInvocation = 'Illegal invocation'; |
| 62 const errIllegalConstructor = 'Illegal constructor'; |
| 63 const errCancelLockedStream = |
| 64 'Cannot cancel a readable stream that is locked to a reader'; |
| 65 const errEnqueueInCloseRequestedStream = |
| 66 'Cannot enqueue a chunk into a readable stream that is closed or has been
requested to be closed'; |
| 67 const errCancelReleasedReader = |
| 68 'This readable stream reader has been released and cannot be used to cance
l its previous owner stream'; |
| 69 const errReadReleasedReader = |
| 70 'This readable stream reader has been released and cannot be used to read
from its previous owner stream'; |
| 71 const errCloseCloseRequestedStream = |
| 72 'Cannot close a readable stream that has already been requested to be clos
ed'; |
| 73 const errCloseErroredStream = 'Cannot close an errored readable stream'; |
| 74 const errErrorClosedStream = 'Cannot error a close readable stream'; |
| 75 const errErrorErroredStream = |
| 76 'Cannot error a readable stream that is already errored'; |
| 77 const errReaderConstructorBadArgument = |
| 78 'ReadableStreamReader constructor argument is not a readable stream'; |
| 79 const errReaderConstructorStreamAlreadyLocked = |
| 80 'ReadableStreamReader constructor can only accept readable streams that ar
e not yet locked to a reader'; |
| 81 const errReleaseReaderWithPendingRead = |
| 82 'Cannot release a readable stream reader when it still has outstanding rea
d() calls that have not yet settled'; |
| 83 const errReleasedReaderClosedPromise = |
| 84 'This readable stream reader has been released and cannot be used to monit
or the stream\'s state'; |
| 85 const errInvalidSize = |
| 86 'The return value of a queuing strategy\'s size function must be a finite,
non-NaN, non-negative number'; |
| 87 const errSizeNotAFunction = |
| 88 'A queuing strategy\'s size property must be a function'; |
| 89 const errInvalidHWM = |
| 90 'A queueing strategy\'s highWaterMark property must be a nonnegative, non-
NaN number'; |
| 91 const errTmplMustBeFunctionOrUndefined = name => |
| 92 `${name} must be a function or undefined`; |
| 93 |
| 94 class ReadableStream { |
| 95 constructor() { |
| 96 // TODO(domenic): when V8 gets default parameters and destructuring, all |
| 97 // this can be cleaned up. |
| 98 const underlyingSource = arguments[0] === undefined ? {} : arguments[0]; |
| 99 const strategy = arguments[1] === undefined ? {} : arguments[1]; |
| 100 const size = strategy.size; |
| 101 let highWaterMark = strategy.highWaterMark; |
| 102 if (highWaterMark === undefined) { |
| 103 highWaterMark = 1; |
| 104 } |
| 105 |
| 106 const normalizedStrategy = |
| 107 ValidateAndNormalizeQueuingStrategy(size, highWaterMark); |
| 108 |
| 109 this[readableStreamUnderlyingSource] = underlyingSource; |
| 110 |
| 111 this[readableStreamQueue] = new v8.InternalPackedArray(); |
| 112 this[readableStreamQueueSize] = 0; |
| 113 |
| 114 this[readableStreamState] = STATE_READABLE; |
| 115 this[readableStreamBits] = 0b0; |
| 116 this[readableStreamReader] = undefined; |
| 117 this[readableStreamStoredError] = undefined; |
| 118 |
| 119 this[readableStreamStrategySize] = normalizedStrategy.size; |
| 120 this[readableStreamStrategyHWM] = normalizedStrategy.highWaterMark; |
| 121 |
| 122 const controller = new ReadableStreamController(this); |
| 123 this[readableStreamController] = controller; |
| 124 |
| 125 const startResult = CallOrNoop( |
| 126 underlyingSource, 'start', controller, 'underlyingSource.start'); |
| 127 thenPromise(Promise_resolve(startResult), |
| 128 () => { |
| 129 this[readableStreamBits] |= STARTED; |
| 130 RequestReadableStreamPull(this); |
| 131 }, |
| 132 r => { |
| 133 if (this[readableStreamState] === STATE_READABLE) { |
| 134 return ErrorReadableStream(this, r); |
| 135 } |
| 136 }); |
| 137 } |
| 138 |
| 139 get locked() { |
| 140 if (IsReadableStream(this) === false) { |
| 141 throw new TypeError(errIllegalInvocation); |
| 142 } |
| 143 |
| 144 return IsReadableStreamLocked(this); |
| 145 } |
| 146 |
| 147 cancel(reason) { |
| 148 if (IsReadableStream(this) === false) { |
| 149 return Promise_reject(new TypeError(errIllegalInvocation)); |
| 150 } |
| 151 |
| 152 if (IsReadableStreamLocked(this) === true) { |
| 153 return Promise_reject(new TypeError(errCancelLockedStream)); |
| 154 } |
| 155 |
| 156 return CancelReadableStream(this, reason); |
| 157 } |
| 158 |
| 159 getReader() { |
| 160 if (IsReadableStream(this) === false) { |
| 161 throw new TypeError(errIllegalInvocation); |
| 162 } |
| 163 |
| 164 return AcquireReadableStreamReader(this); |
| 165 } |
| 166 |
| 167 tee() { |
| 168 if (IsReadableStream(this) === false) { |
| 169 throw new TypeError(errIllegalInvocation); |
| 170 } |
| 171 |
| 172 return TeeReadableStream(this); |
| 173 } |
| 174 } |
| 175 |
| 176 class ReadableStreamController { |
| 177 constructor(stream) { |
| 178 if (IsReadableStream(stream) === false) { |
| 179 throw new TypeError(errIllegalConstructor); |
| 180 } |
| 181 |
| 182 if (stream[readableStreamController] !== undefined) { |
| 183 throw new TypeError(errIllegalConstructor); |
| 184 } |
| 185 |
| 186 this[readableStreamControllerControlledReadableStream] = stream; |
| 187 } |
| 188 |
| 189 get desiredSize() { |
| 190 if (IsReadableStreamController(this) === false) { |
| 191 throw new TypeError(errIllegalInvocation); |
| 192 } |
| 193 |
| 194 return GetReadableStreamDesiredSize( |
| 195 this[readableStreamControllerControlledReadableStream]); |
| 196 } |
| 197 |
| 198 close() { |
| 199 if (IsReadableStreamController(this) === false) { |
| 200 throw new TypeError(errIllegalInvocation); |
| 201 } |
| 202 |
| 203 const stream = this[readableStreamControllerControlledReadableStream]; |
| 204 |
| 205 if (stream[readableStreamBits] & CLOSE_REQUESTED) { |
| 206 throw new TypeError(errCloseCloseRequestedStream); |
| 207 } |
| 208 if (stream[readableStreamState] === STATE_ERRORED) { |
| 209 throw new TypeError(errCloseErroredStream); |
| 210 } |
| 211 |
| 212 return CloseReadableStream(stream); |
| 213 } |
| 214 |
| 215 enqueue(chunk) { |
| 216 if (IsReadableStreamController(this) === false) { |
| 217 throw new TypeError(errIllegalInvocation); |
| 218 } |
| 219 |
| 220 const stream = this[readableStreamControllerControlledReadableStream]; |
| 221 |
| 222 if (stream[readableStreamState] === STATE_ERRORED) { |
| 223 throw stream[readableStreamStoredError]; |
| 224 } |
| 225 |
| 226 if (stream[readableStreamBits] & CLOSE_REQUESTED) { |
| 227 throw new TypeError(errEnqueueInCloseRequestedStream); |
| 228 } |
| 229 |
| 230 return EnqueueInReadableStream(stream, chunk); |
| 231 } |
| 232 |
| 233 error(e) { |
| 234 if (IsReadableStreamController(this) === false) { |
| 235 throw new TypeError(errIllegalInvocation); |
| 236 } |
| 237 |
| 238 const stream = this[readableStreamControllerControlledReadableStream]; |
| 239 |
| 240 const state = stream[readableStreamState]; |
| 241 if (state !== STATE_READABLE) { |
| 242 if (state === STATE_ERRORED) { |
| 243 throw new TypeError(errErrorErroredStream); |
| 244 } |
| 245 if (state === STATE_CLOSED) { |
| 246 throw new TypeError(errErrorClosedStream); |
| 247 } |
| 248 } |
| 249 |
| 250 return ErrorReadableStream(stream, e); |
| 251 } |
| 252 } |
| 253 |
| 254 class ReadableStreamReader { |
| 255 constructor(stream) { |
| 256 if (IsReadableStream(stream) === false) { |
| 257 throw new TypeError(errReaderConstructorBadArgument); |
| 258 } |
| 259 if (IsReadableStreamLocked(stream) === true) { |
| 260 throw new TypeError(errReaderConstructorStreamAlreadyLocked); |
| 261 } |
| 262 |
| 263 this[readableStreamReaderOwnerReadableStream] = stream; |
| 264 stream[readableStreamReader] = this; |
| 265 |
| 266 this[readableStreamReaderReadRequests] = new v8.InternalPackedArray(); |
| 267 |
| 268 switch (stream[readableStreamState]) { |
| 269 case STATE_READABLE: |
| 270 this[readableStreamReaderClosedPromise] = v8.createPromise(); |
| 271 break; |
| 272 case STATE_CLOSED: |
| 273 this[readableStreamReaderClosedPromise] = Promise_resolve(undefined); |
| 274 break; |
| 275 case STATE_ERRORED: |
| 276 this[readableStreamReaderClosedPromise] = |
| 277 Promise_reject(stream[readableStreamStoredError]); |
| 278 break; |
| 279 } |
| 280 } |
| 281 |
| 282 get closed() { |
| 283 if (IsReadableStreamReader(this) === false) { |
| 284 return Promise_reject(new TypeError(errIllegalInvocation)); |
| 285 } |
| 286 |
| 287 return this[readableStreamReaderClosedPromise]; |
| 288 } |
| 289 |
| 290 cancel(reason) { |
| 291 if (IsReadableStreamReader(this) === false) { |
| 292 return Promise_reject(new TypeError(errIllegalInvocation)); |
| 293 } |
| 294 |
| 295 const stream = this[readableStreamReaderOwnerReadableStream]; |
| 296 if (stream === undefined) { |
| 297 return Promise_reject(new TypeError(errCancelReleasedReader)); |
| 298 } |
| 299 |
| 300 return CancelReadableStream(stream, reason); |
| 301 } |
| 302 |
| 303 read() { |
| 304 if (IsReadableStreamReader(this) === false) { |
| 305 return Promise_reject(new TypeError(errIllegalInvocation)); |
| 306 } |
| 307 |
| 308 if (this[readableStreamReaderOwnerReadableStream] === undefined) { |
| 309 return Promise_reject(new TypeError(errReadReleasedReader)); |
| 310 } |
| 311 |
| 312 return ReadFromReadableStreamReader(this); |
| 313 } |
| 314 |
| 315 releaseLock() { |
| 316 if (IsReadableStreamReader(this) === false) { |
| 317 throw new TypeError(errIllegalInvocation); |
| 318 } |
| 319 |
| 320 const stream = this[readableStreamReaderOwnerReadableStream]; |
| 321 if (stream === undefined) { |
| 322 return undefined; |
| 323 } |
| 324 |
| 325 if (this[readableStreamReaderReadRequests].length > 0) { |
| 326 throw new TypeError(errReleaseReaderWithPendingRead); |
| 327 } |
| 328 |
| 329 if (stream[readableStreamState] === STATE_READABLE) { |
| 330 v8.rejectPromise(this[readableStreamReaderClosedPromise], |
| 331 new TypeError(errReleasedReaderClosedPromise)); |
| 332 } else { |
| 333 this[readableStreamReaderClosedPromise] = Promise_reject(new TypeError( |
| 334 errReleasedReaderClosedPromise)); |
| 335 } |
| 336 |
| 337 this[readableStreamReaderOwnerReadableStream][readableStreamReader] = |
| 338 undefined; |
| 339 this[readableStreamReaderOwnerReadableStream] = undefined; |
| 340 } |
| 341 } |
| 342 |
| 343 // |
| 344 // Readable stream abstract operations |
| 345 // |
| 346 |
| 347 function AcquireReadableStreamReader(stream) { |
| 348 return new ReadableStreamReader(stream); |
| 349 } |
| 350 |
| 351 function CancelReadableStream(stream, reason) { |
| 352 stream[readableStreamBits] |= DISTURBED; |
| 353 |
| 354 const state = stream[readableStreamState]; |
| 355 if (state === STATE_CLOSED) { |
| 356 return Promise_resolve(undefined); |
| 357 } |
| 358 if (state === STATE_ERRORED) { |
| 359 return Promise_reject(stream[readableStreamStoredError]); |
| 360 } |
| 361 |
| 362 stream[readableStreamQueue] = new v8.InternalPackedArray(); |
| 363 FinishClosingReadableStream(stream); |
| 364 |
| 365 const underlyingSource = stream[readableStreamUnderlyingSource]; |
| 366 const sourceCancelPromise = PromiseCallOrNoop( |
| 367 underlyingSource, 'cancel', reason, 'underlyingSource.cancel'); |
| 368 return thenPromise(sourceCancelPromise, () => undefined); |
| 369 } |
| 370 |
| 371 function CloseReadableStream(stream) { |
| 372 if (stream[readableStreamState] === STATE_CLOSED) { |
| 373 return undefined; |
| 374 } |
| 375 |
| 376 stream[readableStreamBits] |= CLOSE_REQUESTED; |
| 377 |
| 378 if (stream[readableStreamQueue].length === 0) { |
| 379 return FinishClosingReadableStream(stream); |
| 380 } |
| 381 } |
| 382 |
| 383 function EnqueueInReadableStream(stream, chunk) { |
| 384 if (stream[readableStreamState] === STATE_CLOSED) { |
| 385 return undefined; |
| 386 } |
| 387 |
| 388 if (IsReadableStreamLocked(stream) === true && |
| 389 stream[readableStreamReader][readableStreamReaderReadRequests].length > |
| 390 0) { |
| 391 const readRequest = |
| 392 stream[readableStreamReader][readableStreamReaderReadRequests] |
| 393 .shift(); |
| 394 v8.resolvePromise(readRequest, CreateIterResultObject(chunk, false)); |
| 395 } else { |
| 396 let chunkSize = 1; |
| 397 |
| 398 const strategySize = stream[readableStreamStrategySize]; |
| 399 if (strategySize !== undefined) { |
| 400 try { |
| 401 chunkSize = strategySize(chunk); |
| 402 } catch (chunkSizeE) { |
| 403 ErrorReadableStream(stream, chunkSizeE); |
| 404 throw chunkSizeE; |
| 405 } |
| 406 } |
| 407 |
| 408 try { |
| 409 EnqueueValueWithSize(stream, chunk, chunkSize); |
| 410 } catch (enqueueE) { |
| 411 ErrorReadableStream(stream, enqueueE); |
| 412 throw enqueueE; |
| 413 } |
| 414 } |
| 415 |
| 416 RequestReadableStreamPull(stream); |
| 417 } |
| 418 |
| 419 function ErrorReadableStream(stream, e) { |
| 420 stream[readableStreamQueue] = new v8.InternalPackedArray(); |
| 421 stream[readableStreamStoredError] = e; |
| 422 stream[readableStreamState] = STATE_ERRORED; |
| 423 |
| 424 const reader = stream[readableStreamReader]; |
| 425 if (reader === undefined) { |
| 426 return undefined; |
| 427 } |
| 428 |
| 429 const readRequests = reader[readableStreamReaderReadRequests]; |
| 430 for (let i = 0; i < readRequests.length; ++i) { |
| 431 v8.rejectPromise(readRequests[i], e); |
| 432 } |
| 433 reader[readableStreamReaderReadRequests] = new v8.InternalPackedArray(); |
| 434 |
| 435 v8.rejectPromise(reader[readableStreamReaderClosedPromise], e); |
| 436 } |
| 437 |
| 438 function FinishClosingReadableStream(stream) { |
| 439 stream[readableStreamState] = STATE_CLOSED; |
| 440 |
| 441 const reader = stream[readableStreamReader]; |
| 442 if (reader === undefined) { |
| 443 return undefined; |
| 444 } |
| 445 |
| 446 |
| 447 const readRequests = reader[readableStreamReaderReadRequests]; |
| 448 for (let i = 0; i < readRequests.length; ++i) { |
| 449 v8.resolvePromise( |
| 450 readRequests[i], CreateIterResultObject(undefined, true)); |
| 451 } |
| 452 reader[readableStreamReaderReadRequests] = new v8.InternalPackedArray(); |
| 453 |
| 454 v8.resolvePromise(reader[readableStreamReaderClosedPromise], undefined); |
| 455 } |
| 456 |
| 457 function GetReadableStreamDesiredSize(stream) { |
| 458 const queueSize = GetTotalQueueSize(stream); |
| 459 return stream[readableStreamStrategyHWM] - queueSize; |
| 460 } |
| 461 |
| 462 function IsReadableStream(x) { |
| 463 return hasOwnProperty(x, readableStreamUnderlyingSource); |
| 464 } |
| 465 |
| 466 function IsReadableStreamDisturbed(stream) { |
| 467 return stream[readableStreamBits] & DISTURBED; |
| 468 } |
| 469 |
| 470 function IsReadableStreamLocked(stream) { |
| 471 return stream[readableStreamReader] !== undefined; |
| 472 } |
| 473 |
| 474 function IsReadableStreamController(x) { |
| 475 return hasOwnProperty(x, readableStreamControllerControlledReadableStream); |
| 476 } |
| 477 |
| 478 function IsReadableStreamReader(x) { |
| 479 return hasOwnProperty(x, readableStreamReaderOwnerReadableStream); |
| 480 } |
| 481 |
| 482 function ReadFromReadableStreamReader(reader) { |
| 483 const stream = reader[readableStreamReaderOwnerReadableStream]; |
| 484 stream[readableStreamBits] |= DISTURBED; |
| 485 |
| 486 if (stream[readableStreamState] === STATE_CLOSED) { |
| 487 return Promise_resolve(CreateIterResultObject(undefined, true)); |
| 488 } |
| 489 |
| 490 if (stream[readableStreamState] === STATE_ERRORED) { |
| 491 return Promise_reject(stream[readableStreamStoredError]); |
| 492 } |
| 493 |
| 494 const queue = stream[readableStreamQueue]; |
| 495 if (queue.length > 0) { |
| 496 const chunk = DequeueValue(stream); |
| 497 |
| 498 if (stream[readableStreamBits] & CLOSE_REQUESTED && queue.length === 0) { |
| 499 FinishClosingReadableStream(stream); |
| 500 } else { |
| 501 RequestReadableStreamPull(stream); |
| 502 } |
| 503 |
| 504 return Promise_resolve(CreateIterResultObject(chunk, false)); |
| 505 } else { |
| 506 const readRequest = v8.createPromise(); |
| 507 |
| 508 reader[readableStreamReaderReadRequests].push(readRequest); |
| 509 RequestReadableStreamPull(stream); |
| 510 return readRequest; |
| 511 } |
| 512 } |
| 513 |
| 514 function RequestReadableStreamPull(stream) { |
| 515 const shouldPull = ShouldReadableStreamPull(stream); |
| 516 if (shouldPull === false) { |
| 517 return undefined; |
| 518 } |
| 519 |
| 520 if (stream[readableStreamBits] & PULLING) { |
| 521 stream[readableStreamBits] |= PULL_AGAIN; |
| 522 return undefined; |
| 523 } |
| 524 |
| 525 stream[readableStreamBits] |= PULLING; |
| 526 |
| 527 const underlyingSource = stream[readableStreamUnderlyingSource]; |
| 528 const controller = stream[readableStreamController]; |
| 529 const pullPromise = PromiseCallOrNoop( |
| 530 underlyingSource, 'pull', controller, 'underlyingSource.pull'); |
| 531 |
| 532 thenPromise(pullPromise, |
| 533 () => { |
| 534 stream[readableStreamBits] &= ~PULLING; |
| 535 |
| 536 if (stream[readableStreamBits] & PULL_AGAIN) { |
| 537 stream[readableStreamBits] &= ~PULL_AGAIN; |
| 538 return RequestReadableStreamPull(stream); |
| 539 } |
| 540 }, |
| 541 e => { |
| 542 if (stream[readableStreamState] === STATE_READABLE) { |
| 543 return ErrorReadableStream(stream, e); |
| 544 } |
| 545 }); |
| 546 } |
| 547 |
| 548 function ShouldReadableStreamPull(stream) { |
| 549 const state = stream[readableStreamState]; |
| 550 if (state === STATE_CLOSED || state === STATE_ERRORED) { |
| 551 return false; |
| 552 } |
| 553 |
| 554 if (stream[readableStreamBits] & CLOSE_REQUESTED) { |
| 555 return false; |
| 556 } |
| 557 |
| 558 if (!(stream[readableStreamBits] & STARTED)) { |
| 559 return false; |
| 560 } |
| 561 |
| 562 if (IsReadableStreamLocked(stream) === true) { |
| 563 const reader = stream[readableStreamReader]; |
| 564 const readRequests = reader[readableStreamReaderReadRequests]; |
| 565 if (readRequests.length > 0) { |
| 566 return true; |
| 567 } |
| 568 } |
| 569 |
| 570 const desiredSize = GetReadableStreamDesiredSize(stream); |
| 571 if (desiredSize > 0) { |
| 572 return true; |
| 573 } |
| 574 |
| 575 return false; |
| 576 } |
| 577 |
| 578 // Potential future optimization: use class instances for the underlying |
| 579 // sources, so that we don't re-create |
| 580 // closures every time. |
| 581 |
| 582 // TODO(domenic): shouldClone argument from spec not supported yet |
| 583 function TeeReadableStream(stream) { |
| 584 const reader = AcquireReadableStreamReader(stream); |
| 585 |
| 586 let closedOrErrored = false; |
| 587 let canceled1 = false; |
| 588 let canceled2 = false; |
| 589 let reason1; |
| 590 let reason2; |
| 591 let promise = v8.createPromise(); |
| 592 |
| 593 const branch1 = new ReadableStream({pull, cancel: cancel1}); |
| 594 |
| 595 const branch2 = new ReadableStream({pull, cancel: cancel2}); |
| 596 |
| 597 thenPromise( |
| 598 reader[readableStreamReaderClosedPromise], undefined, function(r) { |
| 599 if (closedOrErrored === true) { |
| 600 return; |
| 601 } |
| 602 |
| 603 ErrorReadableStream(branch1, r); |
| 604 ErrorReadableStream(branch2, r); |
| 605 closedOrErrored = true; |
| 606 }); |
| 607 |
| 608 return [branch1, branch2]; |
| 609 |
| 610 |
| 611 function pull() { |
| 612 return thenPromise( |
| 613 ReadFromReadableStreamReader(reader), function(result) { |
| 614 const value = result.value; |
| 615 const done = result.done; |
| 616 |
| 617 if (done === true && closedOrErrored === false) { |
| 618 CloseReadableStream(branch1); |
| 619 CloseReadableStream(branch2); |
| 620 closedOrErrored = true; |
| 621 } |
| 622 |
| 623 if (closedOrErrored === true) { |
| 624 return; |
| 625 } |
| 626 |
| 627 if (canceled1 === false) { |
| 628 EnqueueInReadableStream(branch1, value); |
| 629 } |
| 630 |
| 631 if (canceled2 === false) { |
| 632 EnqueueInReadableStream(branch2, value); |
| 633 } |
| 634 }); |
| 635 } |
| 636 |
| 637 function cancel1(reason) { |
| 638 canceled1 = true; |
| 639 reason1 = reason; |
| 640 |
| 641 if (canceled2 === true) { |
| 642 const compositeReason = [reason1, reason2]; |
| 643 const cancelResult = CancelReadableStream(stream, compositeReason); |
| 644 v8.resolvePromise(promise, cancelResult); |
| 645 } |
| 646 |
| 647 return promise; |
| 648 } |
| 649 |
| 650 function cancel2(reason) { |
| 651 canceled2 = true; |
| 652 reason2 = reason; |
| 653 |
| 654 if (canceled1 === true) { |
| 655 const compositeReason = [reason1, reason2]; |
| 656 const cancelResult = CancelReadableStream(stream, compositeReason); |
| 657 v8.resolvePromise(promise, cancelResult); |
| 658 } |
| 659 |
| 660 return promise; |
| 661 } |
| 662 } |
| 663 |
| 664 // |
| 665 // Queue-with-sizes |
| 666 // Modified from taking the queue (as in the spec) to taking the stream, so we |
| 667 // can modify the queue size alongside. |
| 668 // |
| 669 |
| 670 function DequeueValue(stream) { |
| 671 const result = stream[readableStreamQueue].shift(); |
| 672 stream[readableStreamQueueSize] -= result.size; |
| 673 return result.value; |
| 674 } |
| 675 |
| 676 function EnqueueValueWithSize(stream, value, size) { |
| 677 size = Number(size); |
| 678 if (Number_isNaN(size) || size === +Infinity || size < 0) { |
| 679 throw new RangeError(errInvalidSize); |
| 680 } |
| 681 |
| 682 stream[readableStreamQueueSize] += size; |
| 683 stream[readableStreamQueue].push({value, size}); |
| 684 } |
| 685 |
| 686 function GetTotalQueueSize(stream) { return stream[readableStreamQueueSize]; } |
| 687 |
| 688 // |
| 689 // Other helpers |
| 690 // |
| 691 |
| 692 function ValidateAndNormalizeQueuingStrategy(size, highWaterMark) { |
| 693 if (size !== undefined && typeof size !== 'function') { |
| 694 throw new TypeError(errSizeNotAFunction); |
| 695 } |
| 696 |
| 697 highWaterMark = Number(highWaterMark); |
| 698 if (Number_isNaN(highWaterMark)) { |
| 699 throw new TypeError(errInvalidHWM); |
| 700 } |
| 701 if (highWaterMark < 0) { |
| 702 throw new RangeError(errInvalidHWM); |
| 703 } |
| 704 |
| 705 return {size, highWaterMark}; |
| 706 } |
| 707 |
| 708 // Modified from InvokeOrNoop in spec |
| 709 function CallOrNoop(O, P, arg, nameForError) { |
| 710 const method = O[P]; |
| 711 if (method === undefined) { |
| 712 return undefined; |
| 713 } |
| 714 if (typeof method !== 'function') { |
| 715 throw new TypeError(errTmplMustBeFunctionOrUndefined(nameForError)); |
| 716 } |
| 717 |
| 718 return callFunction(method, O, arg); |
| 719 } |
| 720 |
| 721 |
| 722 // Modified from PromiseInvokeOrNoop in spec |
| 723 function PromiseCallOrNoop(O, P, arg, nameForError) { |
| 724 let method; |
| 725 try { |
| 726 method = O[P]; |
| 727 } catch (methodE) { |
| 728 return Promise_reject(methodE); |
| 729 } |
| 730 |
| 731 if (method === undefined) { |
| 732 return Promise_resolve(undefined); |
| 733 } |
| 734 |
| 735 if (typeof method !== 'function') { |
| 736 return Promise_reject(errTmplMustBeFunctionOrUndefined(nameForError)); |
| 737 } |
| 738 |
| 739 try { |
| 740 return Promise_resolve(callFunction(method, O, arg)); |
| 741 } catch (e) { |
| 742 return Promise_reject(e); |
| 743 } |
| 744 } |
| 745 |
| 746 function CreateIterResultObject(value, done) { return {value, done}; } |
| 747 |
| 748 |
| 749 // |
| 750 // Additions to the global |
| 751 // |
| 752 |
| 753 defineProperty(global, 'ReadableStream', { |
| 754 value: ReadableStream, |
| 755 enumerable: false, |
| 756 configurable: true, |
| 757 writable: true |
| 758 }); |
| 759 |
| 760 // |
| 761 // Exports to Blink |
| 762 // |
| 763 |
| 764 binding.IsReadableStreamDisturbed = IsReadableStreamDisturbed; |
| 765 }); |
OLD | NEW |