How to use WritableStreamDefaultControllerProcessWrite method in wpt
Best JavaScript code snippet using wpt
writablestream.js
Source:writablestream.js 
1'use strict';2/* eslint-disable no-use-before-define */3const {4 ArrayPrototypePush,5 ArrayPrototypeShift,6 FunctionPrototypeBind,7 FunctionPrototypeCall,8 ObjectDefineProperties,9 PromisePrototypeThen,10 PromiseResolve,11 PromiseReject,12 ReflectConstruct,13 Symbol,14 SymbolToStringTag,15} = primordials;16const {17 codes: {18 ERR_ILLEGAL_CONSTRUCTOR,19 ERR_INVALID_ARG_VALUE,20 ERR_INVALID_ARG_TYPE,21 ERR_INVALID_STATE,22 ERR_INVALID_THIS,23 },24} = require('internal/errors');25const {26 DOMException,27} = internalBinding('messaging');28const {29 createDeferredPromise,30 customInspectSymbol: kInspect,31} = require('internal/util');32const {33 MessageChannel,34} = require('internal/worker/io');35const {36 kDeserialize,37 kTransfer,38 kTransferList,39 makeTransferable,40} = require('internal/worker/js_transferable');41const {42 customInspect,43 dequeueValue,44 ensureIsPromise,45 enqueueValueWithSize,46 extractHighWaterMark,47 extractSizeAlgorithm,48 lazyTransfer,49 isBrandCheck,50 isPromisePending,51 peekQueueValue,52 resetQueue,53 setPromiseHandled,54 nonOpCancel,55 nonOpStart,56 nonOpWrite,57 kType,58 kState,59 kEnumerableProperty,60} = require('internal/webstreams/util');61const {62 AbortController,63} = require('internal/abort_controller');64const assert = require('internal/assert');65const kAbort = Symbol('kAbort');66const kCloseSentinel = Symbol('kCloseSentinel');67const kError = Symbol('kError');68/**69 * @typedef {import('../abort_controller').AbortSignal} AbortSignal70 * @typedef {import('./queuingstrategies').QueuingStrategy71 * } QueuingStrategy72 * @typedef {import('./queuingstrategies').QueuingStrategySize73 * } QueuingStrategySize74 *75 * @callback UnderlyingSinkStartCallback76 * @param {WritableStreamDefaultController} controller77 *78 * @callback UnderlyingSinkWriteCallback79 * @param {any} chunk80 * @param {WritableStreamDefaultController} controller81 * @returns {Promise<void>}82 *83 * @callback UnderlyingSinkCloseCallback84 * @returns {Promise<void>}85 *86 * @callback UnderlyingSinkAbortCallback87 * @param {any} reason88 * @returns {Promise<void>}89 *90 * @typedef {{91 * start? : UnderlyingSinkStartCallback,92 * write? : UnderlyingSinkWriteCallback,93 * close? : UnderlyingSinkCloseCallback,94 * abort? : UnderlyingSinkAbortCallback,95 * type? : any,96 * }} UnderlyingSink97 */98class WritableStream {99 [kType] = 'WritableStream';100 get [SymbolToStringTag]() { return this[kType]; }101 /**102 * @param {UnderlyingSink} [sink]103 * @param {QueuingStrategy} [strategy]104 */105 constructor(sink = null, strategy = {}) {106 const type = sink?.type;107 if (type !== undefined)108 throw new ERR_INVALID_ARG_VALUE.RangeError('type', type);109 this[kState] = {110 close: createDeferredPromise(),111 closeRequest: {112 promise: undefined,113 resolve: undefined,114 reject: undefined,115 },116 inFlightWriteRequest: {117 promise: undefined,118 resolve: undefined,119 reject: undefined,120 },121 inFlightCloseRequest: {122 promise: undefined,123 resolve: undefined,124 reject: undefined,125 },126 pendingAbortRequest: {127 abort: {128 promise: undefined,129 resolve: undefined,130 reject: undefined,131 },132 reason: undefined,133 wasAlreadyErroring: false,134 },135 backpressure: false,136 controller: undefined,137 state: 'writable',138 storedError: undefined,139 writeRequests: [],140 writer: undefined,141 transfer: {142 readable: undefined,143 port1: undefined,144 port2: undefined,145 promise: undefined,146 }147 };148 const size = extractSizeAlgorithm(strategy?.size);149 const highWaterMark = extractHighWaterMark(strategy?.highWaterMark, 1);150 setupWritableStreamDefaultControllerFromSink(151 this,152 sink,153 highWaterMark,154 size);155 // eslint-disable-next-line no-constructor-return156 return makeTransferable(this);157 }158 /**159 * @readonly160 * @type {boolean}161 */162 get locked() {163 if (!isWritableStream(this))164 throw new ERR_INVALID_THIS('WritableStream');165 return isWritableStreamLocked(this);166 }167 /**168 * @param {any} reason169 * @returns {Promise<vois>}170 */171 abort(reason = undefined) {172 if (!isWritableStream(this))173 return PromiseReject(new ERR_INVALID_THIS('WritableStream'));174 if (isWritableStreamLocked(this)) {175 return PromiseReject(176 new ERR_INVALID_STATE.TypeError('WritableStream is locked'));177 }178 return writableStreamAbort(this, reason);179 }180 /**181 * @returns {Promise<void>}182 */183 close() {184 if (!isWritableStream(this))185 return PromiseReject(new ERR_INVALID_THIS('WritableStream'));186 if (isWritableStreamLocked(this)) {187 return PromiseReject(188 new ERR_INVALID_STATE.TypeError('WritableStream is locked'));189 }190 if (writableStreamCloseQueuedOrInFlight(this)) {191 return PromiseReject(192 new ERR_INVALID_STATE.TypeError('Failure closing WritableStream'));193 }194 return writableStreamClose(this);195 }196 /**197 * @returns {WritableStreamDefaultWriter}198 */199 getWriter() {200 if (!isWritableStream(this))201 throw new ERR_INVALID_THIS('WritableStream');202 return new WritableStreamDefaultWriter(this);203 }204 [kInspect](depth, options) {205 return customInspect(depth, options, this[kType], {206 locked: this.locked,207 state: this[kState].state,208 });209 }210 [kTransfer]() {211 if (!isWritableStream(this))212 throw new ERR_INVALID_THIS('WritableStream');213 if (this.locked) {214 this[kState].transfer.port1?.close();215 this[kState].transfer.port1 = undefined;216 this[kState].transfer.port2 = undefined;217 throw new DOMException(218 'Cannot transfer a locked WritableStream',219 'DataCloneError');220 }221 const {222 readable,223 promise,224 } = lazyTransfer().newCrossRealmReadableStream(225 this,226 this[kState].transfer.port1);227 this[kState].transfer.readable = readable;228 this[kState].transfer.promise = promise;229 setPromiseHandled(this[kState].transfer.promise);230 return {231 data: { port: this[kState].transfer.port2 },232 deserializeInfo:233 'internal/webstreams/writablestream:TransferredWritableStream'234 };235 }236 [kTransferList]() {237 const { port1, port2 } = new MessageChannel();238 this[kState].transfer.port1 = port1;239 this[kState].transfer.port2 = port2;240 return [ port2 ];241 }242 [kDeserialize]({ port }) {243 const transfer = lazyTransfer();244 setupWritableStreamDefaultControllerFromSink(245 this,246 new transfer.CrossRealmTransformWritableSink(port),247 1,248 () => 1);249 }250}251ObjectDefineProperties(WritableStream.prototype, {252 locked: kEnumerableProperty,253 abort: kEnumerableProperty,254 close: kEnumerableProperty,255 getWriter: kEnumerableProperty,256});257function TransferredWritableStream() {258 return makeTransferable(ReflectConstruct(259 function() {260 this[kType] = 'WritableStream';261 this[kState] = {262 close: createDeferredPromise(),263 closeRequest: {264 promise: undefined,265 resolve: undefined,266 reject: undefined,267 },268 inFlightWriteRequest: {269 promise: undefined,270 resolve: undefined,271 reject: undefined,272 },273 inFlightCloseRequest: {274 promise: undefined,275 resolve: undefined,276 reject: undefined,277 },278 pendingAbortRequest: {279 abort: {280 promise: undefined,281 resolve: undefined,282 reject: undefined,283 },284 reason: undefined,285 wasAlreadyErroring: false,286 },287 backpressure: false,288 controller: undefined,289 state: 'writable',290 storedError: undefined,291 writeRequests: [],292 writer: undefined,293 transfer: {294 promise: undefined,295 port1: undefined,296 port2: undefined,297 readable: undefined,298 },299 };300 },301 [], WritableStream));302}303TransferredWritableStream.prototype[kDeserialize] = () => {};304class WritableStreamDefaultWriter {305 [kType] = 'WritableStreamDefaultWriter';306 get [SymbolToStringTag]() { return this[kType]; }307 /**308 * @param {WritableStream} stream309 */310 constructor(stream) {311 if (!isWritableStream(stream))312 throw new ERR_INVALID_ARG_TYPE('stream', 'WritableStream', stream);313 this[kState] = {314 stream: undefined,315 close: {316 promise: undefined,317 resolve: undefined,318 reject: undefined,319 },320 ready: {321 promise: undefined,322 resolve: undefined,323 reject: undefined,324 }325 };326 setupWritableStreamDefaultWriter(this, stream);327 }328 /**329 * @readonly330 * @type {Promise<void>}331 */332 get closed() {333 if (!isWritableStreamDefaultWriter(this))334 return PromiseReject(new ERR_INVALID_THIS('WritableStreamDefaultWriter'));335 return this[kState].close.promise;336 }337 /**338 * @readonly339 * @type {number}340 */341 get desiredSize() {342 if (!isWritableStreamDefaultWriter(this))343 throw new ERR_INVALID_THIS('WritableStreamDefaultWriter');344 if (this[kState].stream === undefined) {345 throw new ERR_INVALID_STATE.TypeError(346 'Writer is not bound to a WritableStream');347 }348 return writableStreamDefaultWriterGetDesiredSize(this);349 }350 /**351 * @readonly352 * @type {Promise<void>}353 */354 get ready() {355 if (!isWritableStreamDefaultWriter(this))356 return PromiseReject(new ERR_INVALID_THIS('WritableStreamDefaultWriter'));357 return this[kState].ready.promise;358 }359 /**360 * @param {any} reason361 * @returns {Promise<void>}362 */363 abort(reason = undefined) {364 if (!isWritableStreamDefaultWriter(this))365 return PromiseReject(new ERR_INVALID_THIS('WritableStreamDefaultWriter'));366 if (this[kState].stream === undefined) {367 return PromiseReject(368 new ERR_INVALID_STATE.TypeError(369 'Writer is not bound to a WritableStream'));370 }371 return writableStreamDefaultWriterAbort(this, reason);372 }373 /**374 * @returns {Promise<void>}375 */376 close() {377 if (!isWritableStreamDefaultWriter(this))378 return PromiseReject(new ERR_INVALID_THIS('WritableStreamDefaultWriter'));379 const {380 stream,381 } = this[kState];382 if (stream === undefined) {383 return PromiseReject(384 new ERR_INVALID_STATE.TypeError(385 'Writer is not bound to a WritableStream'));386 }387 if (writableStreamCloseQueuedOrInFlight(stream)) {388 return PromiseReject(389 new ERR_INVALID_STATE.TypeError('Failure to close WritableStream'));390 }391 return writableStreamDefaultWriterClose(this);392 }393 releaseLock() {394 if (!isWritableStreamDefaultWriter(this))395 throw new ERR_INVALID_THIS('WritableStreamDefaultWriter');396 const {397 stream,398 } = this[kState];399 if (stream === undefined)400 return;401 assert(stream[kState].writer !== undefined);402 writableStreamDefaultWriterRelease(this);403 }404 /**405 * @param {any} chunk406 * @returns {Promise<void>}407 */408 write(chunk = undefined) {409 if (!isWritableStreamDefaultWriter(this))410 return PromiseReject(new ERR_INVALID_THIS('WritableStreamDefaultWriter'));411 if (this[kState].stream === undefined) {412 return PromiseReject(413 new ERR_INVALID_STATE.TypeError(414 'Writer is not bound to a WritableStream'));415 }416 return writableStreamDefaultWriterWrite(this, chunk);417 }418 [kInspect](depth, options) {419 return customInspect(depth, options, this[kType], {420 stream: this[kState].stream,421 close: this[kState].close.promise,422 ready: this[kState].ready.promise,423 desiredSize: this.desiredSize,424 });425 }426}427ObjectDefineProperties(WritableStreamDefaultWriter.prototype, {428 closed: kEnumerableProperty,429 ready: kEnumerableProperty,430 desiredSize: kEnumerableProperty,431 abort: kEnumerableProperty,432 close: kEnumerableProperty,433 releaseLock: kEnumerableProperty,434 write: kEnumerableProperty,435});436class WritableStreamDefaultController {437 [kType] = 'WritableStreamDefaultController';438 get [SymbolToStringTag]() { return this[kType]; }439 constructor() {440 throw new ERR_ILLEGAL_CONSTRUCTOR();441 }442 [kAbort](reason) {443 const result = this[kState].abortAlgorithm(reason);444 writableStreamDefaultControllerClearAlgorithms(this);445 return result;446 }447 [kError]() {448 resetQueue(this);449 }450 /**451 * @type {any}452 */453 get abortReason() {454 if (!isWritableStreamDefaultController(this))455 throw new ERR_INVALID_THIS('WritableStreamDefaultController');456 return this[kState].abortReason;457 }458 /**459 * @type {AbortSignal}460 */461 get signal() {462 if (!isWritableStreamDefaultController(this))463 throw new ERR_INVALID_THIS('WritableStreamDefaultController');464 return this[kState].abortController.signal;465 }466 /**467 * @param {any} error468 */469 error(error = undefined) {470 if (!isWritableStreamDefaultController(this))471 throw new ERR_INVALID_THIS('WritableStreamDefaultController');472 if (this[kState].stream[kState].state !== 'writable')473 return;474 writableStreamDefaultControllerError(this, error);475 }476 [kInspect](depth, options) {477 return customInspect(depth, options, this[kType], {478 stream: this[kState].stream,479 });480 }481}482ObjectDefineProperties(WritableStreamDefaultController.prototype, {483 abortReason: kEnumerableProperty,484 signal: kEnumerableProperty,485 error: kEnumerableProperty,486});487function createWritableStreamDefaultController() {488 return ReflectConstruct(489 function() {490 this[kType] = 'WritableStreamDefaultController';491 },492 [], WritableStreamDefaultController);493}494const isWritableStream =495 isBrandCheck('WritableStream');496const isWritableStreamDefaultWriter =497 isBrandCheck('WritableStreamDefaultWriter');498const isWritableStreamDefaultController =499 isBrandCheck('WritableStreamDefaultController');500function isWritableStreamLocked(stream) {501 return stream[kState].writer !== undefined;502}503function setupWritableStreamDefaultWriter(writer, stream) {504 if (isWritableStreamLocked(stream))505 throw new ERR_INVALID_STATE.TypeError('WritableStream is locked');506 writer[kState].stream = stream;507 stream[kState].writer = writer;508 switch (stream[kState].state) {509 case 'writable':510 if (!writableStreamCloseQueuedOrInFlight(stream) &&511 stream[kState].backpressure) {512 writer[kState].ready = createDeferredPromise();513 } else {514 writer[kState].ready = {515 promise: PromiseResolve(),516 resolve: undefined,517 reject: undefined,518 };519 }520 setClosedPromiseToNewPromise();521 break;522 case 'erroring':523 writer[kState].ready = {524 promise: PromiseReject(stream[kState].storedError),525 resolve: undefined,526 reject: undefined,527 };528 setPromiseHandled(writer[kState].ready.promise);529 setClosedPromiseToNewPromise();530 break;531 case 'closed':532 writer[kState].ready = {533 promise: PromiseResolve(),534 resolve: undefined,535 reject: undefined,536 };537 writer[kState].close = {538 promise: PromiseResolve(),539 resolve: undefined,540 reject: undefined,541 };542 break;543 default:544 writer[kState].ready = {545 promise: PromiseReject(stream[kState].storedError),546 resolve: undefined,547 reject: undefined,548 };549 writer[kState].close = {550 promise: PromiseReject(stream[kState].storedError),551 resolve: undefined,552 reject: undefined,553 };554 setPromiseHandled(writer[kState].ready.promise);555 setPromiseHandled(writer[kState].close.promise);556 }557 function setClosedPromiseToNewPromise() {558 writer[kState].close = createDeferredPromise();559 }560}561function writableStreamAbort(stream, reason) {562 const {563 state,564 controller,565 } = stream[kState];566 if (state === 'closed' || state === 'errored')567 return PromiseResolve();568 controller[kState].abortReason = reason;569 controller[kState].abortController.abort();570 if (stream[kState].pendingAbortRequest.abort.promise !== undefined)571 return stream[kState].pendingAbortRequest.abort.promise;572 assert(state === 'writable' || state === 'erroring');573 let wasAlreadyErroring = false;574 if (state === 'erroring') {575 wasAlreadyErroring = true;576 reason = undefined;577 }578 const abort = createDeferredPromise();579 stream[kState].pendingAbortRequest = {580 abort,581 reason,582 wasAlreadyErroring,583 };584 if (!wasAlreadyErroring)585 writableStreamStartErroring(stream, reason);586 return abort.promise;587}588function writableStreamClose(stream) {589 const {590 state,591 writer,592 backpressure,593 controller,594 } = stream[kState];595 if (state === 'closed' || state === 'errored') {596 return PromiseReject(597 new ERR_INVALID_STATE.TypeError('WritableStream is closed'));598 }599 assert(state === 'writable' || state === 'erroring');600 assert(!writableStreamCloseQueuedOrInFlight(stream));601 stream[kState].closeRequest = createDeferredPromise();602 const { promise } = stream[kState].closeRequest;603 if (writer !== undefined && backpressure && state === 'writable')604 writer[kState].ready.resolve?.();605 writableStreamDefaultControllerClose(controller);606 return promise;607}608function writableStreamUpdateBackpressure(stream, backpressure) {609 assert(stream[kState].state === 'writable');610 assert(!writableStreamCloseQueuedOrInFlight(stream));611 const {612 writer,613 } = stream[kState];614 if (writer !== undefined && stream[kState].backpressure !== backpressure) {615 if (backpressure) {616 writer[kState].ready = createDeferredPromise();617 } else {618 writer[kState].ready.resolve?.();619 }620 }621 stream[kState].backpressure = backpressure;622}623function writableStreamStartErroring(stream, reason) {624 assert(stream[kState].storedError === undefined);625 assert(stream[kState].state === 'writable');626 const {627 controller,628 writer,629 } = stream[kState];630 assert(controller !== undefined);631 stream[kState].state = 'erroring';632 stream[kState].storedError = reason;633 if (writer !== undefined) {634 writableStreamDefaultWriterEnsureReadyPromiseRejected(writer, reason);635 }636 if (!writableStreamHasOperationMarkedInFlight(stream) &&637 controller[kState].started) {638 writableStreamFinishErroring(stream);639 }640}641function writableStreamRejectCloseAndClosedPromiseIfNeeded(stream) {642 assert(stream[kState].state === 'errored');643 if (stream[kState].closeRequest.promise !== undefined) {644 assert(stream[kState].inFlightCloseRequest.promise === undefined);645 stream[kState].closeRequest.reject?.(stream[kState].storedError);646 stream[kState].closeRequest = {647 promise: undefined,648 reject: undefined,649 resolve: undefined,650 };651 }652 const {653 writer,654 } = stream[kState];655 if (writer !== undefined) {656 writer[kState].close.reject?.(stream[kState].storedError);657 setPromiseHandled(writer[kState].close.promise);658 }659}660function writableStreamMarkFirstWriteRequestInFlight(stream) {661 assert(stream[kState].inFlightWriteRequest.promise === undefined);662 assert(stream[kState].writeRequests.length);663 const writeRequest = ArrayPrototypeShift(stream[kState].writeRequests);664 stream[kState].inFlightWriteRequest = writeRequest;665}666function writableStreamMarkCloseRequestInFlight(stream) {667 assert(stream[kState].inFlightWriteRequest.promise === undefined);668 assert(stream[kState].closeRequest.promise !== undefined);669 stream[kState].inFlightCloseRequest = stream[kState].closeRequest;670 stream[kState].closeRequest = {671 promise: undefined,672 resolve: undefined,673 reject: undefined,674 };675}676function writableStreamHasOperationMarkedInFlight(stream) {677 const {678 inFlightWriteRequest,679 inFlightCloseRequest,680 } = stream[kState];681 if (inFlightWriteRequest.promise === undefined &&682 inFlightCloseRequest.promise === undefined) {683 return false;684 }685 return true;686}687function writableStreamFinishInFlightWriteWithError(stream, error) {688 assert(stream[kState].inFlightWriteRequest.promise !== undefined);689 stream[kState].inFlightWriteRequest.reject?.(error);690 stream[kState].inFlightWriteRequest = {691 promise: undefined,692 resolve: undefined,693 reject: undefined,694 };695 assert(stream[kState].state === 'writable' ||696 stream[kState].state === 'erroring');697 writableStreamDealWithRejection(stream, error);698}699function writableStreamFinishInFlightWrite(stream) {700 assert(stream[kState].inFlightWriteRequest.promise !== undefined);701 stream[kState].inFlightWriteRequest.resolve?.();702 stream[kState].inFlightWriteRequest = {703 promise: undefined,704 resolve: undefined,705 reject: undefined,706 };707}708function writableStreamFinishInFlightCloseWithError(stream, error) {709 assert(stream[kState].inFlightCloseRequest.promise !== undefined);710 stream[kState].inFlightCloseRequest.reject?.(error);711 stream[kState].inFlightCloseRequest = {712 promise: undefined,713 resolve: undefined,714 reject: undefined,715 };716 assert(stream[kState].state === 'writable' ||717 stream[kState].state === 'erroring');718 if (stream[kState].pendingAbortRequest.abort.promise !== undefined) {719 stream[kState].pendingAbortRequest.abort.reject?.(error);720 stream[kState].pendingAbortRequest = {721 abort: {722 promise: undefined,723 resolve: undefined,724 reject: undefined,725 },726 reason: undefined,727 wasAlreadyErroring: false,728 };729 }730 writableStreamDealWithRejection(stream, error);731}732function writableStreamFinishInFlightClose(stream) {733 assert(stream[kState].inFlightCloseRequest.promise !== undefined);734 stream[kState].inFlightCloseRequest.resolve?.();735 stream[kState].inFlightCloseRequest = {736 promise: undefined,737 resolve: undefined,738 reject: undefined,739 };740 if (stream[kState].state === 'erroring') {741 stream[kState].storedError = undefined;742 if (stream[kState].pendingAbortRequest.abort.promise !== undefined) {743 stream[kState].pendingAbortRequest.abort.resolve?.();744 stream[kState].pendingAbortRequest = {745 abort: {746 promise: undefined,747 resolve: undefined,748 reject: undefined,749 },750 reason: undefined,751 wasAlreadyErroring: false,752 };753 }754 }755 stream[kState].state = 'closed';756 if (stream[kState].writer !== undefined)757 stream[kState].writer[kState].close.resolve?.();758 assert(stream[kState].pendingAbortRequest.abort.promise === undefined);759 assert(stream[kState].storedError === undefined);760}761function writableStreamFinishErroring(stream) {762 assert(stream[kState].state === 'erroring');763 assert(!writableStreamHasOperationMarkedInFlight(stream));764 stream[kState].state = 'errored';765 stream[kState].controller[kError]();766 const storedError = stream[kState].storedError;767 for (let n = 0; n < stream[kState].writeRequests.length; n++)768 stream[kState].writeRequests[n].reject?.(storedError);769 stream[kState].writeRequests = [];770 if (stream[kState].pendingAbortRequest.abort.promise === undefined) {771 writableStreamRejectCloseAndClosedPromiseIfNeeded(stream);772 return;773 }774 const abortRequest = stream[kState].pendingAbortRequest;775 stream[kState].pendingAbortRequest = {776 abort: {777 promise: undefined,778 resolve: undefined,779 reject: undefined,780 },781 reason: undefined,782 wasAlreadyErroring: false,783 };784 if (abortRequest.wasAlreadyErroring) {785 abortRequest.abort.reject?.(storedError);786 writableStreamRejectCloseAndClosedPromiseIfNeeded(stream);787 return;788 }789 PromisePrototypeThen(790 ensureIsPromise(791 stream[kState].controller[kAbort],792 stream[kState].controller,793 abortRequest.reason),794 () => {795 abortRequest.abort.resolve?.();796 writableStreamRejectCloseAndClosedPromiseIfNeeded(stream);797 },798 (error) => {799 abortRequest.abort.reject?.(error);800 writableStreamRejectCloseAndClosedPromiseIfNeeded(stream);801 });802}803function writableStreamDealWithRejection(stream, error) {804 const {805 state,806 } = stream[kState];807 if (state === 'writable') {808 writableStreamStartErroring(stream, error);809 return;810 }811 assert(state === 'erroring');812 writableStreamFinishErroring(stream);813}814function writableStreamCloseQueuedOrInFlight(stream) {815 if (stream[kState].closeRequest.promise === undefined &&816 stream[kState].inFlightCloseRequest.promise === undefined) {817 return false;818 }819 return true;820}821function writableStreamAddWriteRequest(stream) {822 assert(isWritableStreamLocked(stream));823 assert(stream[kState].state === 'writable');824 const {825 promise,826 resolve,827 reject,828 } = createDeferredPromise();829 ArrayPrototypePush(830 stream[kState].writeRequests,831 {832 promise,833 resolve,834 reject,835 });836 return promise;837}838function writableStreamDefaultWriterWrite(writer, chunk) {839 const {840 stream,841 } = writer[kState];842 assert(stream !== undefined);843 const {844 controller,845 } = stream[kState];846 const chunkSize = writableStreamDefaultControllerGetChunkSize(847 controller,848 chunk);849 if (stream !== writer[kState].stream) {850 return PromiseReject(851 new ERR_INVALID_STATE.TypeError('Mismatched WritableStreams'));852 }853 const {854 state,855 } = stream[kState];856 if (state === 'errored')857 return PromiseReject(stream[kState].storedError);858 if (writableStreamCloseQueuedOrInFlight(stream) || state === 'closed') {859 return PromiseReject(860 new ERR_INVALID_STATE.TypeError('WritableStream is closed'));861 }862 if (state === 'erroring')863 return PromiseReject(stream[kState].storedError);864 assert(state === 'writable');865 const promise = writableStreamAddWriteRequest(stream);866 writableStreamDefaultControllerWrite(controller, chunk, chunkSize);867 return promise;868}869function writableStreamDefaultWriterRelease(writer) {870 const {871 stream,872 } = writer[kState];873 assert(stream !== undefined);874 assert(stream[kState].writer === writer);875 const releasedError =876 new ERR_INVALID_STATE.TypeError('Writer has been released');877 writableStreamDefaultWriterEnsureReadyPromiseRejected(writer, releasedError);878 writableStreamDefaultWriterEnsureClosedPromiseRejected(writer, releasedError);879 stream[kState].writer = undefined;880 writer[kState].stream = undefined;881}882function writableStreamDefaultWriterGetDesiredSize(writer) {883 const {884 stream,885 } = writer[kState];886 switch (stream[kState].state) {887 case 'errored':888 // Fall through889 case 'erroring':890 return null;891 case 'closed':892 return 0;893 }894 return writableStreamDefaultControllerGetDesiredSize(895 stream[kState].controller);896}897function writableStreamDefaultWriterEnsureReadyPromiseRejected(writer, error) {898 if (isPromisePending(writer[kState].ready.promise)) {899 writer[kState].ready.reject?.(error);900 } else {901 writer[kState].ready = {902 promise: PromiseReject(error),903 resolve: undefined,904 reject: undefined,905 };906 }907 setPromiseHandled(writer[kState].ready.promise);908}909function writableStreamDefaultWriterEnsureClosedPromiseRejected(writer, error) {910 if (isPromisePending(writer[kState].close.promise)) {911 writer[kState].close.reject?.(error);912 } else {913 writer[kState].close = {914 promise: PromiseReject(error),915 resolve: undefined,916 reject: undefined,917 };918 }919 setPromiseHandled(writer[kState].close.promise);920}921function writableStreamDefaultWriterCloseWithErrorPropagation(writer) {922 const {923 stream,924 } = writer[kState];925 assert(stream !== undefined);926 const {927 state,928 } = stream[kState];929 if (writableStreamCloseQueuedOrInFlight(stream) || state === 'closed')930 return PromiseResolve();931 if (state === 'errored')932 return PromiseReject(stream[kState].storedError);933 assert(state === 'writable' || state === 'erroring');934 return writableStreamDefaultWriterClose(writer);935}936function writableStreamDefaultWriterClose(writer) {937 const {938 stream,939 } = writer[kState];940 assert(stream !== undefined);941 return writableStreamClose(stream);942}943function writableStreamDefaultWriterAbort(writer, reason) {944 const {945 stream,946 } = writer[kState];947 assert(stream !== undefined);948 return writableStreamAbort(stream, reason);949}950function writableStreamDefaultControllerWrite(controller, chunk, chunkSize) {951 try {952 enqueueValueWithSize(controller, chunk, chunkSize);953 } catch (error) {954 writableStreamDefaultControllerErrorIfNeeded(controller, error);955 return;956 }957 const {958 stream,959 } = controller[kState];960 if (!writableStreamCloseQueuedOrInFlight(stream) &&961 stream[kState].state === 'writable') {962 writableStreamUpdateBackpressure(963 stream,964 writableStreamDefaultControllerGetBackpressure(controller));965 }966 writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);967}968function writableStreamDefaultControllerProcessWrite(controller, chunk) {969 const {970 stream,971 writeAlgorithm,972 } = controller[kState];973 writableStreamMarkFirstWriteRequestInFlight(stream);974 PromisePrototypeThen(975 ensureIsPromise(writeAlgorithm, controller, chunk, controller),976 () => {977 writableStreamFinishInFlightWrite(stream);978 const {979 state,980 } = stream[kState];981 assert(state === 'writable' || state === 'erroring');982 dequeueValue(controller);983 if (!writableStreamCloseQueuedOrInFlight(stream) &&984 state === 'writable') {985 writableStreamUpdateBackpressure(986 stream,987 writableStreamDefaultControllerGetBackpressure(controller));988 }989 writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);990 },991 (error) => {992 if (stream[kState].state === 'writable')993 writableStreamDefaultControllerClearAlgorithms(controller);994 writableStreamFinishInFlightWriteWithError(stream, error);995 });996}997function writableStreamDefaultControllerProcessClose(controller) {998 const {999 closeAlgorithm,1000 queue,1001 stream,1002 } = controller[kState];1003 writableStreamMarkCloseRequestInFlight(stream);1004 dequeueValue(controller);1005 assert(!queue.length);1006 const sinkClosePromise = ensureIsPromise(closeAlgorithm, controller);1007 writableStreamDefaultControllerClearAlgorithms(controller);1008 PromisePrototypeThen(1009 sinkClosePromise,1010 () => writableStreamFinishInFlightClose(stream),1011 (error) => writableStreamFinishInFlightCloseWithError(stream, error));1012}1013function writableStreamDefaultControllerGetDesiredSize(controller) {1014 const {1015 highWaterMark,1016 queueTotalSize,1017 } = controller[kState];1018 return highWaterMark - queueTotalSize;1019}1020function writableStreamDefaultControllerGetChunkSize(controller, chunk) {1021 try {1022 return FunctionPrototypeCall(1023 controller[kState].sizeAlgorithm,1024 undefined,1025 chunk);1026 } catch (error) {1027 writableStreamDefaultControllerErrorIfNeeded(controller, error);1028 return 1;1029 }1030}1031function writableStreamDefaultControllerErrorIfNeeded(controller, error) {1032 const {1033 stream,1034 } = controller[kState];1035 if (stream[kState].state === 'writable')1036 writableStreamDefaultControllerError(controller, error);1037}1038function writableStreamDefaultControllerError(controller, error) {1039 const {1040 stream,1041 } = controller[kState];1042 assert(stream[kState].state === 'writable');1043 writableStreamDefaultControllerClearAlgorithms(controller);1044 writableStreamStartErroring(stream, error);1045}1046function writableStreamDefaultControllerClose(controller) {1047 enqueueValueWithSize(controller, kCloseSentinel, 0);1048 writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);1049}1050function writableStreamDefaultControllerClearAlgorithms(controller) {1051 controller[kState].writeAlgorithm = undefined;1052 controller[kState].closeAlgorithm = undefined;1053 controller[kState].abortAlgorithm = undefined;1054 controller[kState].sizeAlgorithm = undefined;1055}1056function writableStreamDefaultControllerGetBackpressure(controller) {1057 return writableStreamDefaultControllerGetDesiredSize(controller) <= 0;1058}1059function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) {1060 const {1061 queue,1062 started,1063 stream,1064 } = controller[kState];1065 if (!started || stream[kState].inFlightWriteRequest.promise !== undefined)1066 return;1067 if (stream[kState].state === 'erroring') {1068 writableStreamFinishErroring(stream);1069 return;1070 }1071 if (!queue.length)1072 return;1073 const value = peekQueueValue(controller);1074 if (value === kCloseSentinel)1075 writableStreamDefaultControllerProcessClose(controller);1076 else1077 writableStreamDefaultControllerProcessWrite(controller, value);1078}1079function setupWritableStreamDefaultControllerFromSink(1080 stream,1081 sink,1082 highWaterMark,1083 sizeAlgorithm) {1084 const controller = createWritableStreamDefaultController();1085 const start = sink?.start;1086 const write = sink?.write;1087 const close = sink?.close;1088 const abort = sink?.abort;1089 const startAlgorithm = start ?1090 FunctionPrototypeBind(start, sink, controller) :1091 nonOpStart;1092 const writeAlgorithm = write ?1093 FunctionPrototypeBind(write, sink) :1094 nonOpWrite;1095 const closeAlgorithm = close ?1096 FunctionPrototypeBind(close, sink) : nonOpCancel;1097 const abortAlgorithm = abort ?1098 FunctionPrototypeBind(abort, sink) : nonOpCancel;1099 setupWritableStreamDefaultController(1100 stream,1101 controller,1102 startAlgorithm,1103 writeAlgorithm,1104 closeAlgorithm,1105 abortAlgorithm,1106 highWaterMark,1107 sizeAlgorithm);1108}1109function setupWritableStreamDefaultController(1110 stream,1111 controller,1112 startAlgorithm,1113 writeAlgorithm,1114 closeAlgorithm,1115 abortAlgorithm,1116 highWaterMark,1117 sizeAlgorithm) {1118 assert(isWritableStream(stream));1119 assert(stream[kState].controller === undefined);1120 controller[kState] = {1121 abortAlgorithm,1122 abortReason: undefined,1123 closeAlgorithm,1124 highWaterMark,1125 queue: [],1126 queueTotalSize: 0,1127 abortController: new AbortController(),1128 sizeAlgorithm,1129 started: false,1130 stream,1131 writeAlgorithm,1132 };1133 stream[kState].controller = controller;1134 writableStreamUpdateBackpressure(1135 stream,1136 writableStreamDefaultControllerGetBackpressure(controller));1137 const startResult = startAlgorithm();1138 PromisePrototypeThen(1139 PromiseResolve(startResult),1140 () => {1141 assert(stream[kState].state === 'writable' ||1142 stream[kState].state === 'erroring');1143 controller[kState].started = true;1144 writableStreamDefaultControllerAdvanceQueueIfNeeded(controller);1145 },1146 (error) => {1147 assert(stream[kState].state === 'writable' ||1148 stream[kState].state === 'erroring');1149 controller[kState].started = true;1150 writableStreamDealWithRejection(stream, error);1151 });1152}1153module.exports = {1154 WritableStream,1155 WritableStreamDefaultWriter,1156 WritableStreamDefaultController,1157 TransferredWritableStream,1158 // Exported Brand Checks1159 isWritableStream,1160 isWritableStreamDefaultController,1161 isWritableStreamDefaultWriter,1162 isWritableStreamLocked,1163 setupWritableStreamDefaultWriter,1164 writableStreamAbort,1165 writableStreamClose,1166 writableStreamUpdateBackpressure,1167 writableStreamStartErroring,1168 writableStreamRejectCloseAndClosedPromiseIfNeeded,1169 writableStreamMarkFirstWriteRequestInFlight,1170 writableStreamMarkCloseRequestInFlight,1171 writableStreamHasOperationMarkedInFlight,1172 writableStreamFinishInFlightWriteWithError,1173 writableStreamFinishInFlightWrite,1174 writableStreamFinishInFlightCloseWithError,1175 writableStreamFinishInFlightClose,1176 writableStreamFinishErroring,1177 writableStreamDealWithRejection,1178 writableStreamCloseQueuedOrInFlight,1179 writableStreamAddWriteRequest,1180 writableStreamDefaultWriterWrite,1181 writableStreamDefaultWriterRelease,1182 writableStreamDefaultWriterGetDesiredSize,1183 writableStreamDefaultWriterEnsureReadyPromiseRejected,1184 writableStreamDefaultWriterEnsureClosedPromiseRejected,1185 writableStreamDefaultWriterCloseWithErrorPropagation,1186 writableStreamDefaultWriterClose,1187 writableStreamDefaultWriterAbort,1188 writableStreamDefaultControllerWrite,1189 writableStreamDefaultControllerProcessWrite,1190 writableStreamDefaultControllerProcessClose,1191 writableStreamDefaultControllerGetDesiredSize,1192 writableStreamDefaultControllerGetChunkSize,1193 writableStreamDefaultControllerErrorIfNeeded,1194 writableStreamDefaultControllerError,1195 writableStreamDefaultControllerClose,1196 writableStreamDefaultControllerClearAlgorithms,1197 writableStreamDefaultControllerGetBackpressure,1198 writableStreamDefaultControllerAdvanceQueueIfNeeded,1199 setupWritableStreamDefaultControllerFromSink,1200 setupWritableStreamDefaultController,1201};...
writable_stream_controller.ts
Source:writable_stream_controller.ts
...234 const writeRecord = PeekQueueValue(controller);235 if (writeRecord === "close") {236 WritableStreamDefaultControllerProcessClose(controller);237 } else {238 WritableStreamDefaultControllerProcessWrite(controller, writeRecord.chunk);239 }240}241export function WritableStreamDefaultControllerErrorIfNeeded<T>(242 controller: WritableStreamDefaultController<T>,243 error244) {245 if (controller.controlledWritableStream.state === "writable") {246 WritableStreamDefaultControllerError(controller, error);247 }248}249export function WritableStreamDefaultControllerProcessClose<T>(250 controller: WritableStreamDefaultController<T>251) {252 const stream = controller.controlledWritableStream;...
Using AI Code Generation
1test(() => {2 const ws = new WritableStream();3 const writer = ws.getWriter();4 const writerClosed = writer.closed;5 writer.close();6 return writerClosed;7}, 'WritableStream with no underlying sink: writer.closed resolves immediately');
Using AI Code Generation
1var WritableStream = require('stream').Writable;2var ws = new WritableStream({3 write(chunk, encoding, callback) {4 callback();5 }6});7var writer = ws.getWriter();8writer.write('a');9writer.write('b');10writer.write('c');11writer.releaseLock();12writer.write('d');13writer.write('e');14writer.write('f');15writer.close();16writer.write('g');17writer.write('h');18writer.write('i');
Using AI Code Generation
1var ws = new WritableStream({2 write: function(chunk, controller) {3 controller.processWrite(chunk);4 }5});6ws.write("a");7ws.write("b");8ws.write("c");9var Writable = require('stream').Writable;10var ws = new Writable({11 write: function(chunk, encoding, callback) {12 this._writableState.controller.processWrite(chunk);13 callback();14 }15});16ws.write("a");17ws.write("b");18ws.write("c");
Using AI Code Generation
1const { WritableStream, WritableStreamDefaultControllerProcessWrite } = require('stream/web');2const { Writable } = require('stream');3const { TextEncoder } = require('util');4const { once } = require('events');5const { strictEqual } = require('assert');6const { PassThrough } = require('stream');7const writer = new WritableStream({8 write(chunk) {9 console.log(chunk);10 }11}).getWriter();12(async () => {13 const controller = await writer.getController();14 const chunk = new TextEncoder().encode('test');15 WritableStreamDefaultControllerProcessWrite(controller, chunk);16 await once(writer, 'close');17})();18const { Writable } = require('stream');19const { TextEncoder } = require('util');20const { once } = require('events');21const { strictEqual } = require('assert');22const { PassThrough } = require('stream');23const writer = new Writable({24 write(chunk, encoding, callback) {25 console.log(chunk);26 callback();27 }28});29const chunk = new TextEncoder().encode('test');30writer.write(chunk);31writer.end();32const { WritableStream, WritableStreamDefaultControllerProcessWrite } = require('stream/web');33const { TextEncoder } = require('util');34const writer = new WritableStream({35 write(chunk) {36 console.log(chunk);37 }38}).getWriter();39(async () => {40 const controller = await writer.getController();41 const chunk = new TextEncoder().encode('test');
Using AI Code Generation
1var writer = writable.getWriter();2writer.write("a");3writer.write("b");4writer.write("c");5writer.close();6test(function() {7 var rs = new ReadableStream({8 pull: function(c) {9 c.enqueue("a");10 c.enqueue("b");11 c.enqueue("c");12 c.close();13 }14 });15 var ws = new WritableStream({16 write: function(chunk) {17 assert_equals(chunk, "a");18 return Promise.resolve();19 }20 });21 return rs.pipeTo(ws);22}, "WritableStreamDefaultControllerProcessWrite");23function test(func, name, properties) {24 var test = async_test(name, properties);25 test.step(function() { func(test); });26}
Using AI Code Generation
1var WritableStreamDefaultControllerProcessWrite = function (2) {3 var stream = controller._controlledWritableStream;4 var state = stream._state;5 if (state === "writable") {6 var writer = stream._writer;7 var chunkSize = 1;8 var writeRecord = {9 };10 try {11 var result = writer._writeAlgorithm(writeRecord);12 } catch (writeE) {13 WritableStreamDefaultControllerErrorIfNeeded(controller, writeE);14 return;15 }16 WritableStreamDefaultControllerUpdateBackpressure(17 );18 }19};20var WritableStreamDefaultControllerUpdateBackpressure = function (21) {22 var stream = controller._controlledWritableStream;23 var state = stream._state;24 if (state === "writable") {25 WritableStreamDefaultControllerGetBackpressure(controller);26 WritableStreamDefaultControllerUpdateBackpressureIfNeeded(controller);27 if (backpressure && !controller._backpressure) {28 var writer = stream._writer;29 if (writer._readyPromise_resolve !== undefined) {30 writer._readyPromise_resolve(undefined);31 writer._readyPromise_resolve = undefined;32 writer._readyPromise_reject = undefined;33 }34 } else if (!backpressure && controller._backpressure) {35 var writer = stream._writer;36 if (writer._readyPromise_resolve === undefined) {37 writer._readyPromise_resolve = new Promise(function (resolve, reject) {38 writer._readyPromise_resolve = resolve;39 writer._readyPromise_reject = reject;40 });41 }42 }43 }44};45var WritableStreamDefaultControllerUpdateBackpressureIfNeeded = function (46) {47 var stream = controller._controlledWritableStream;48 var state = stream._state;49 if (state === "writable") {50 WritableStreamDefaultControllerGetBackpressure(controller);51 if (backpressure !== controller._backpressure) {
Using AI Code Generation
1var ws = new WritableStream({2 write(chunk, controller) {3 return WritableStreamDefaultControllerProcessWrite(controller, chunk);4 }5});6var writer = ws.getWriter();7writer.write('a');8var ws = new WritableStream({9 write(chunk, controller) {10 return WritableStreamDefaultControllerProcessWrite(controller, chunk);11 }12});13var writer = ws.getWriter();14writer.write('a');15var ws = new WritableStream({16 write(chunk, controller) {17 return WritableStreamDefaultControllerProcessWrite(controller, chunk);18 }19});20var writer = ws.getWriter();21writer.write('a');22var ws = new WritableStream({23 write(chunk, controller) {24 return WritableStreamDefaultControllerProcessWrite(controller, chunk);25 }26});27var writer = ws.getWriter();28writer.write('a');29var ws = new WritableStream({30 write(chunk, controller) {31 return WritableStreamDefaultControllerProcessWrite(controller, chunk);32 }33});34var writer = ws.getWriter();35writer.write('a');36var ws = new WritableStream({37 write(chunk, controller) {38 return WritableStreamDefaultControllerProcessWrite(controller, chunk);39 }40});41var writer = ws.getWriter();42writer.write('a');43var ws = new WritableStream({44 write(chunk, controller) {45 return WritableStreamDefaultControllerProcessWrite(controller, chunk);46 }47});48var writer = ws.getWriter();49writer.write('a');50var ws = new WritableStream({51 write(chunk, controller) {52 return WritableStreamDefaultControllerProcessWrite(controller, chunk);53 }54});55var writer = ws.getWriter();56writer.write('a');
Using AI Code Generation
1const assert = require('assert');2const { ReadableStream, WritableStream, CountQueuingStrategy } = require('stream/web');3const rs = new ReadableStream({4 start(controller) {5 controller.enqueue('a');6 controller.enqueue('b');7 controller.enqueue('c');8 controller.close();9 },10});11const ws = new WritableStream({12 write(chunk) {13 return new Promise(resolve => {14 setTimeout(() => {15 console.log(chunk);16 resolve();17 }, 100);18 });19 },20 writeAborted: {21 },22});23const writer = ws.getWriter();24const reader = rs.getReader();25writer.write('a');26writer.write('b');27writer.write('c');28reader.read().then(console.log);29reader.read().then(console.log);30reader.read().then(console.log);31writer.abort();32reader.read().then(console.log);33reader.read().then(console.log);34reader.read().then(console.log);35writer.close();36reader.read().then(console.log);37reader.read().then(console.log);38reader.read().then(console.log);39{ value: 'a', done: false }40{ value: 'b', done: false }41{ value:
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run wpt automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
