Skip to main content

script/dom/stream/
writablestreamdefaultcontroller.rs

1/* This Source Code Form is subject to the terms of the Mozilla Public
2 * License, v. 2.0. If a copy of the MPL was not distributed with this
3 * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
4
5use std::cell::{Cell, RefCell};
6use std::ptr;
7use std::rc::Rc;
8
9use dom_struct::dom_struct;
10use js::context::JSContext;
11use js::jsapi::{Heap, JSObject};
12use js::jsval::{JSVal, UndefinedValue};
13use js::realm::CurrentRealm;
14use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
15use script_bindings::reflector::{Reflector, reflect_dom_object_with_cx};
16
17use crate::dom::bindings::callback::ExceptionHandling;
18use crate::dom::bindings::codegen::Bindings::QueuingStrategyBinding::QueuingStrategySize;
19use crate::dom::bindings::codegen::Bindings::UnderlyingSinkBinding::{
20    UnderlyingSinkAbortCallback, UnderlyingSinkCloseCallback, UnderlyingSinkStartCallback,
21    UnderlyingSinkWriteCallback,
22};
23use crate::dom::bindings::codegen::Bindings::WritableStreamDefaultControllerBinding::WritableStreamDefaultControllerMethods;
24use crate::dom::bindings::error::{Error, ErrorToJsval, Fallible};
25use crate::dom::bindings::reflector::DomGlobal;
26use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
27use crate::dom::globalscope::GlobalScope;
28use crate::dom::messageport::MessagePort;
29use crate::dom::promise::{Promise, RootedPromise, TracedPromise};
30use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
31use crate::dom::readablestreamdefaultcontroller::{EnqueuedValue, QueueWithSizes, ValueWithSize};
32use crate::dom::stream::writablestream::WritableStream;
33use crate::dom::types::{AbortController, AbortSignal, TransformStream};
34use crate::realms::enter_auto_realm;
35
36impl js::gc::Rootable for CloseAlgorithmFulfillmentHandler {}
37
38/// The fulfillment handler for
39/// <https://streams.spec.whatwg.org/#writable-stream-default-controller-process-close>
40#[derive(Clone, JSTraceable, MallocSizeOf)]
41#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
42struct CloseAlgorithmFulfillmentHandler {
43    stream: Dom<WritableStream>,
44}
45
46impl Callback for CloseAlgorithmFulfillmentHandler {
47    fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
48        let stream = self.stream.as_rooted();
49
50        // Perform ! WritableStreamFinishInFlightClose(stream).
51        stream.finish_in_flight_close(cx);
52    }
53}
54
55impl js::gc::Rootable for CloseAlgorithmRejectionHandler {}
56
57/// The rejection handler for
58/// <https://streams.spec.whatwg.org/#writable-stream-default-controller-process-close>
59#[derive(Clone, JSTraceable, MallocSizeOf)]
60#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
61struct CloseAlgorithmRejectionHandler {
62    stream: Dom<WritableStream>,
63}
64
65impl Callback for CloseAlgorithmRejectionHandler {
66    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
67        let stream = self.stream.as_rooted();
68
69        let global = GlobalScope::from_current_realm(cx);
70
71        // Perform ! WritableStreamFinishInFlightCloseWithError(stream, reason).
72        stream.finish_in_flight_close_with_error(cx, &global, v);
73    }
74}
75
76impl js::gc::Rootable for StartAlgorithmFulfillmentHandler {}
77
78/// The fulfillment handler for
79/// <https://streams.spec.whatwg.org/#set-up-writable-stream-default-controller>
80#[derive(Clone, JSTraceable, MallocSizeOf)]
81#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
82struct StartAlgorithmFulfillmentHandler {
83    controller: Dom<WritableStreamDefaultController>,
84}
85
86impl Callback for StartAlgorithmFulfillmentHandler {
87    /// Continuation of <https://streams.spec.whatwg.org/#set-up-writable-stream-default-controller>
88    /// Upon fulfillment of startPromise,
89    fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
90        let controller = self.controller.as_rooted();
91        let stream = controller
92            .stream
93            .get()
94            .expect("Controller should have a stream.");
95
96        // Assert: stream.[[state]] is "writable" or "erroring".
97        assert!(stream.is_erroring() || stream.is_writable());
98
99        // Set controller.[[started]] to true.
100        controller.started.set(true);
101
102        let global = GlobalScope::from_current_realm(cx);
103
104        // Perform ! WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller).
105        controller.advance_queue_if_needed(cx, &global)
106    }
107}
108
109impl js::gc::Rootable for StartAlgorithmRejectionHandler {}
110
111/// The rejection handler for
112/// <https://streams.spec.whatwg.org/#set-up-writable-stream-default-controller>
113#[derive(Clone, JSTraceable, MallocSizeOf)]
114#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
115struct StartAlgorithmRejectionHandler {
116    controller: Dom<WritableStreamDefaultController>,
117}
118
119impl Callback for StartAlgorithmRejectionHandler {
120    /// Continuation of <https://streams.spec.whatwg.org/#set-up-writable-stream-default-controller>
121    /// Upon rejection of startPromise with reason r,
122    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
123        let controller = self.controller.as_rooted();
124        let stream = controller
125            .stream
126            .get()
127            .expect("Controller should have a stream.");
128
129        // Assert: stream.[[state]] is "writable" or "erroring".
130        assert!(stream.is_erroring() || stream.is_writable());
131
132        // Set controller.[[started]] to true.
133        controller.started.set(true);
134
135        let global = GlobalScope::from_current_realm(cx);
136
137        // Perform ! WritableStreamDealWithRejection(stream, r).
138        stream.deal_with_rejection(cx, &global, v);
139    }
140}
141
142impl js::gc::Rootable for TransferBackPressurePromiseReaction {}
143
144/// Reacting to backpressurePromise as part of the `writeAlgorithm` of
145/// <https://streams.spec.whatwg.org/#abstract-opdef-setupcrossrealmtransformwritable>
146#[derive(JSTraceable, MallocSizeOf)]
147#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
148struct TransferBackPressurePromiseReaction {
149    /// The result of reacting to backpressurePromise.
150    result_promise: TracedPromise,
151
152    /// The backpressurePromise.
153    #[ignore_malloc_size_of = "nested Rc"]
154    backpressure_promise: Rc<RefCell<Option<TracedPromise>>>,
155
156    /// The chunk received by the `writeAlgorithm`.
157    #[ignore_malloc_size_of = "mozjs"]
158    chunk: Box<Heap<JSVal>>,
159
160    /// The port used in the algorithm.
161    port: Dom<MessagePort>,
162}
163
164impl Callback for TransferBackPressurePromiseReaction {
165    /// Reacting to backpressurePromise with the following fulfillment steps:
166    fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
167        let global = self.result_promise.global();
168        // Set backpressurePromise to a new promise.
169        let promise = Promise::new_rooted(cx, &global);
170        *self.backpressure_promise.borrow_mut() = Some(promise.to_traced());
171
172        // Let result be PackAndPostMessageHandlingError(port, "chunk", chunk).
173        rooted!(&in(cx) let mut chunk = UndefinedValue());
174        chunk.set(self.chunk.get());
175        let result = self
176            .port
177            .pack_and_post_message_handling_error(cx, "chunk", chunk.handle());
178
179        // If result is an abrupt completion,
180        if let Err(error) = result {
181            // Disentangle port.
182            global.disentangle_port(cx, &self.port);
183
184            // Return a promise rejected with result.[[Value]].
185            self.result_promise.reject_error(cx, error);
186        } else {
187            // Otherwise, return a promise resolved with undefined.
188            self.result_promise.resolve_native(cx, &());
189        }
190    }
191}
192
193impl js::gc::Rootable for WriteAlgorithmFulfillmentHandler {}
194
195/// The fulfillment handler for
196/// <https://streams.spec.whatwg.org/#writable-stream-default-controller-process-write>
197#[derive(Clone, JSTraceable, MallocSizeOf)]
198#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
199struct WriteAlgorithmFulfillmentHandler {
200    controller: Dom<WritableStreamDefaultController>,
201}
202
203impl Callback for WriteAlgorithmFulfillmentHandler {
204    fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
205        let controller = self.controller.as_rooted();
206        let stream = controller
207            .stream
208            .get()
209            .expect("Controller should have a stream.");
210
211        // Perform ! WritableStreamFinishInFlightWrite(stream).
212        stream.finish_in_flight_write(cx);
213
214        // Let state be stream.[[state]].
215        // Assert: state is "writable" or "erroring".
216        assert!(stream.is_erroring() || stream.is_writable());
217
218        // Perform ! DequeueValue(controller).
219        rooted!(&in(cx) let mut rval = UndefinedValue());
220        controller.queue.dequeue_value(cx, Some(rval.handle_mut()));
221
222        let global = GlobalScope::from_current_realm(cx);
223
224        // If ! WritableStreamCloseQueuedOrInFlight(stream) is false and state is "writable",
225        if !stream.close_queued_or_in_flight() && stream.is_writable() {
226            // Let backpressure be ! WritableStreamDefaultControllerGetBackpressure(controller).
227            let backpressure = controller.get_backpressure();
228
229            // Perform ! WritableStreamUpdateBackpressure(stream, backpressure).
230            stream.update_backpressure(cx, backpressure, &global);
231        }
232
233        // Perform ! WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller).
234        controller.advance_queue_if_needed(cx, &global)
235    }
236}
237
238impl js::gc::Rootable for WriteAlgorithmRejectionHandler {}
239
240/// The rejection handler for
241/// <https://streams.spec.whatwg.org/#writable-stream-default-controller-process-write>
242#[derive(Clone, JSTraceable, MallocSizeOf)]
243#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
244struct WriteAlgorithmRejectionHandler {
245    controller: Dom<WritableStreamDefaultController>,
246}
247
248impl Callback for WriteAlgorithmRejectionHandler {
249    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
250        let controller = self.controller.as_rooted();
251        let stream = controller
252            .stream
253            .get()
254            .expect("Controller should have a stream.");
255
256        // If stream.[[state]] is "writable",
257        if stream.is_writable() {
258            // perform ! WritableStreamDefaultControllerClearAlgorithms(controller).
259            controller.clear_algorithms();
260        }
261
262        let global = GlobalScope::from_current_realm(cx);
263
264        // Perform ! WritableStreamFinishInFlightWriteWithError(stream, reason).
265        stream.finish_in_flight_write_with_error(cx, &global, v);
266    }
267}
268
269/// The type of sink algorithms we are using.
270#[derive(JSTraceable, PartialEq)]
271#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
272pub enum UnderlyingSinkType {
273    /// Algorithms are provided by Js callbacks.
274    Js {
275        /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-abortalgorithm>
276        abort: RefCell<Option<Rc<UnderlyingSinkAbortCallback>>>,
277
278        start: RefCell<Option<Rc<UnderlyingSinkStartCallback>>>,
279
280        /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-closealgorithm>
281        close: RefCell<Option<Rc<UnderlyingSinkCloseCallback>>>,
282
283        /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-writealgorithm>
284        write: RefCell<Option<Rc<UnderlyingSinkWriteCallback>>>,
285    },
286    /// Algorithms supporting streams transfer are implemented in Rust.
287    /// The promise and port used in those algorithms are stored here.
288    Transfer {
289        backpressure_promise: Rc<RefCell<Option<TracedPromise>>>,
290        port: Dom<MessagePort>,
291    },
292    /// Algorithms supporting transform streams are implemented in Rust.
293    Transform(Dom<TransformStream>, TracedPromise),
294}
295
296impl UnderlyingSinkType {
297    pub(crate) fn new_js(
298        abort: Option<Rc<UnderlyingSinkAbortCallback>>,
299        start: Option<Rc<UnderlyingSinkStartCallback>>,
300        close: Option<Rc<UnderlyingSinkCloseCallback>>,
301        write: Option<Rc<UnderlyingSinkWriteCallback>>,
302    ) -> Self {
303        UnderlyingSinkType::Js {
304            abort: RefCell::new(abort),
305            start: RefCell::new(start),
306            close: RefCell::new(close),
307            write: RefCell::new(write),
308        }
309    }
310}
311
312/// <https://streams.spec.whatwg.org/#ws-default-controller-class>
313#[dom_struct]
314pub struct WritableStreamDefaultController {
315    reflector_: Reflector,
316
317    /// The type of underlying sink used. Besides the default JS one,
318    /// there will be others for stream transfer, and for transform stream.
319    #[ignore_malloc_size_of = "underlying_sink_type"]
320    underlying_sink_type: UnderlyingSinkType,
321
322    /// The JS object used as `this` when invoking sink algorithms.
323    #[ignore_malloc_size_of = "mozjs"]
324    underlying_sink_obj: Heap<*mut JSObject>,
325
326    /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-queue>
327    queue: QueueWithSizes,
328
329    /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-started>
330    started: Cell<bool>,
331
332    /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-strategyhwm>
333    strategy_hwm: f64,
334
335    /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-strategysizealgorithm>
336    #[ignore_malloc_size_of = "QueuingStrategySize"]
337    strategy_size: RefCell<Option<Rc<QueuingStrategySize>>>,
338
339    /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-stream>
340    stream: MutNullableDom<WritableStream>,
341
342    /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-abortcontroller>
343    abort_controller: Dom<AbortController>,
344}
345
346impl WritableStreamDefaultController {
347    /// <https://streams.spec.whatwg.org/#set-up-writable-stream-default-controller-from-underlying-sink>
348    #[cfg_attr(crown, expect(crown::unrooted_must_root))]
349    fn new_inherited(
350        cx: &mut JSContext,
351        global: &GlobalScope,
352        underlying_sink_type: UnderlyingSinkType,
353        strategy_hwm: f64,
354        strategy_size: Rc<QueuingStrategySize>,
355    ) -> WritableStreamDefaultController {
356        WritableStreamDefaultController {
357            reflector_: Reflector::new(),
358            underlying_sink_type,
359            queue: Default::default(),
360            stream: Default::default(),
361            underlying_sink_obj: Default::default(),
362            strategy_hwm,
363            strategy_size: RefCell::new(Some(strategy_size)),
364            started: Default::default(),
365            abort_controller: Dom::from_ref(&AbortController::new_with_proto(cx, global, None)),
366        }
367    }
368
369    #[cfg_attr(crown, expect(crown::unrooted_must_root))]
370    pub(crate) fn new(
371        cx: &mut JSContext,
372        global: &GlobalScope,
373        underlying_sink_type: UnderlyingSinkType,
374        strategy_hwm: f64,
375        strategy_size: Rc<QueuingStrategySize>,
376    ) -> DomRoot<WritableStreamDefaultController> {
377        reflect_dom_object_with_cx(
378            Box::new(WritableStreamDefaultController::new_inherited(
379                cx,
380                global,
381                underlying_sink_type,
382                strategy_hwm,
383                strategy_size,
384            )),
385            global,
386            cx,
387        )
388    }
389
390    pub(crate) fn started(&self) -> bool {
391        self.started.get()
392    }
393
394    /// Setting the JS object after the heap has settled down.
395    pub(crate) fn set_underlying_sink_this_object(&self, this_object: SafeHandleObject) {
396        self.underlying_sink_obj.set(*this_object);
397    }
398
399    /// "Signal abort" call from <https://streams.spec.whatwg.org/#writable-stream-abort>
400    pub(crate) fn signal_abort(&self, cx: &mut CurrentRealm, reason: SafeHandleValue) {
401        self.abort_controller.signal_abort(cx, reason);
402    }
403
404    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-clear-algorithms>
405    fn clear_algorithms(&self) {
406        match &self.underlying_sink_type {
407            UnderlyingSinkType::Js {
408                abort,
409                start: _,
410                close,
411                write,
412            } => {
413                // Set controller.[[writeAlgorithm]] to undefined.
414                write.borrow_mut().take();
415
416                // Set controller.[[closeAlgorithm]] to undefined.
417                close.borrow_mut().take();
418
419                // Set controller.[[abortAlgorithm]] to undefined.
420                abort.borrow_mut().take();
421            },
422            UnderlyingSinkType::Transfer {
423                backpressure_promise,
424                ..
425            } => {
426                backpressure_promise.borrow_mut().take();
427            },
428            UnderlyingSinkType::Transform(_, _) => {
429                return;
430            },
431        }
432
433        // Set controller.[[strategySizeAlgorithm]] to undefined.
434        self.strategy_size.borrow_mut().take();
435    }
436
437    /// <https://streams.spec.whatwg.org/#set-up-writable-stream-default-controller>
438    pub(crate) fn setup(
439        &self,
440        cx: &mut JSContext,
441        global: &GlobalScope,
442        stream: &WritableStream,
443    ) -> Result<(), Error> {
444        // Assert: stream implements WritableStream.
445        // Implied by stream type.
446
447        // Assert: stream.[[controller]] is undefined.
448        stream.assert_no_controller();
449
450        // Set controller.[[stream]] to stream.
451        self.stream.set(Some(stream));
452
453        // Set stream.[[controller]] to controller.
454        stream.set_default_controller(self);
455
456        // Perform ! ResetQueue(controller).
457
458        // Set controller.[[abortController]] to a new AbortController.
459
460        // Set controller.[[started]] to false.
461
462        // Set controller.[[strategySizeAlgorithm]] to sizeAlgorithm.
463
464        // Set controller.[[strategyHWM]] to highWaterMark.
465
466        // Set controller.[[writeAlgorithm]] to writeAlgorithm.
467
468        // Set controller.[[closeAlgorithm]] to closeAlgorithm.
469
470        // Set controller.[[abortAlgorithm]] to abortAlgorithm.
471
472        // Note: above steps are done in `new_inherited`.
473
474        // Let backpressure be ! WritableStreamDefaultControllerGetBackpressure(controller).
475        let backpressure = self.get_backpressure();
476
477        // Perform ! WritableStreamUpdateBackpressure(stream, backpressure).
478        stream.update_backpressure(cx, backpressure, global);
479
480        // Let startResult be the result of performing startAlgorithm. (This may throw an exception.)
481        // Let startPromise be a promise resolved with startResult.
482        let start_promise = self.start_algorithm(cx, global)?;
483
484        let rooted_default_controller = DomRoot::from_ref(self);
485
486        // Upon fulfillment of startPromise,
487        rooted!(&in(cx) let mut fulfillment_handler = Some(StartAlgorithmFulfillmentHandler {
488            controller: Dom::from_ref(&rooted_default_controller),
489        }));
490
491        // Upon rejection of startPromise with reason r,
492        rooted!(&in(cx) let mut rejection_handler = Some(StartAlgorithmRejectionHandler {
493            controller: Dom::from_ref(&rooted_default_controller),
494        }));
495
496        let handler = PromiseNativeHandler::new(
497            cx,
498            global,
499            fulfillment_handler.take().map(|h| Box::new(h) as Box<_>),
500            rejection_handler.take().map(|h| Box::new(h) as Box<_>),
501        );
502        let mut realm = enter_auto_realm(cx, global);
503        let cx = &mut realm.current_realm();
504        start_promise.append_native_handler(cx, &handler);
505
506        Ok(())
507    }
508
509    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-close>
510    pub(crate) fn close(&self, cx: &mut JSContext, global: &GlobalScope) {
511        // Perform ! EnqueueValueWithSize(controller, close sentinel, 0).
512        self.queue
513            .enqueue_value_with_size(EnqueuedValue::CloseSentinel)
514            .expect("Enqueuing the close sentinel should not fail.");
515        // Perform ! WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller).
516        self.advance_queue_if_needed(cx, global);
517    }
518
519    fn start_algorithm(&self, cx: &mut JSContext, global: &GlobalScope) -> Fallible<RootedPromise> {
520        match &self.underlying_sink_type {
521            UnderlyingSinkType::Js {
522                start,
523                abort: _,
524                close: _,
525                write: _,
526            } => {
527                let algo = start.borrow().clone();
528                let start_promise = if let Some(start) = algo {
529                    rooted!(&in(cx) let mut result: JSVal);
530                    rooted!(&in(cx) let this_object = self.underlying_sink_obj.get());
531                    start.Call_(
532                        cx,
533                        &this_object.handle(),
534                        self,
535                        result.handle_mut(),
536                        ExceptionHandling::Rethrow,
537                    )?;
538                    Promise::resolve_or_wrap_promise(cx, result.handle(), global)
539                } else {
540                    // Let startAlgorithm be an algorithm that returns undefined.
541                    Promise::new_resolved_rooted(cx, global, ())
542                };
543
544                Ok(start_promise)
545            },
546            UnderlyingSinkType::Transfer { .. } => {
547                // Let startAlgorithm be an algorithm that returns undefined.
548                Ok(Promise::new_resolved_rooted(cx, global, ()))
549            },
550            UnderlyingSinkType::Transform(_, start_promise) => {
551                // Let startAlgorithm be an algorithm that returns startPromise.
552                Ok(start_promise.root(cx))
553            },
554        }
555    }
556
557    /// <https://streams.spec.whatwg.org/#ref-for-abstract-opdef-writablestreamcontroller-abortsteps>
558    pub(crate) fn abort_steps(
559        &self,
560        cx: &mut JSContext,
561        global: &GlobalScope,
562        reason: SafeHandleValue,
563    ) -> RootedPromise {
564        let result = match &self.underlying_sink_type {
565            UnderlyingSinkType::Js {
566                abort,
567                start: _,
568                close: _,
569                write: _,
570            } => {
571                rooted!(&in(cx) let this_object = self.underlying_sink_obj.get());
572                let algo = abort.borrow().clone();
573                // Let result be the result of performing this.[[abortAlgorithm]], passing reason.
574                let result = if let Some(algo) = algo {
575                    algo.Call_(
576                        cx,
577                        &this_object.handle(),
578                        Some(reason),
579                        ExceptionHandling::Rethrow,
580                    )
581                } else {
582                    Ok(Promise::new_resolved_rooted(cx, global, ()))
583                };
584                result.unwrap_or_else(|e| {
585                    let promise = Promise::new_rooted(cx, global);
586                    promise.reject_error(cx, e);
587                    promise
588                })
589            },
590            UnderlyingSinkType::Transfer { port, .. } => {
591                // The steps from the `abortAlgorithm` at
592                // <https://streams.spec.whatwg.org/#abstract-opdef-setupcrossrealmtransformwritable>
593
594                // Let result be PackAndPostMessageHandlingError(port, "error", reason).
595                let result = port.pack_and_post_message_handling_error(cx, "error", reason);
596
597                // Disentangle port.
598                global.disentangle_port(cx, port);
599
600                let promise = Promise::new_rooted(cx, global);
601
602                // If result is an abrupt completion, return a promise rejected with result.[[Value]]
603                if let Err(error) = result {
604                    promise.reject_error(cx, error);
605                } else {
606                    // Otherwise, return a promise resolved with undefined.
607                    promise.resolve_native(cx, &());
608                }
609                promise
610            },
611            UnderlyingSinkType::Transform(stream, _) => {
612                // Return ! TransformStreamDefaultSinkAbortAlgorithm(stream, reason).
613                stream
614                    .transform_stream_default_sink_abort_algorithm(cx, global, reason)
615                    .expect("Transform stream default sink abort algorithm should not fail.")
616            },
617        };
618
619        // Perform ! WritableStreamDefaultControllerClearAlgorithms(controller).
620        self.clear_algorithms();
621
622        result
623    }
624
625    /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-writealgorithm>
626    fn call_write_algorithm(
627        &self,
628        cx: &mut JSContext,
629        chunk: SafeHandleValue,
630        global: &GlobalScope,
631    ) -> RootedPromise {
632        match &self.underlying_sink_type {
633            UnderlyingSinkType::Js {
634                abort: _,
635                start: _,
636                close: _,
637                write,
638            } => {
639                rooted!(&in(cx) let this_object = self.underlying_sink_obj.get());
640                let algo = write.borrow().clone();
641                let result = if let Some(algo) = algo {
642                    algo.Call_(
643                        cx,
644                        &this_object.handle(),
645                        chunk,
646                        self,
647                        ExceptionHandling::Rethrow,
648                    )
649                } else {
650                    Ok(Promise::new_resolved_rooted(cx, global, ()))
651                };
652                result.unwrap_or_else(|e| {
653                    let promise = Promise::new_rooted(cx, global);
654                    promise.reject_error(cx, e);
655                    promise
656                })
657            },
658            UnderlyingSinkType::Transfer {
659                backpressure_promise,
660                port,
661            } => {
662                // The steps from the `writeAlgorithm` at
663                // <https://streams.spec.whatwg.org/#abstract-opdef-setupcrossrealmtransformwritable>
664
665                // If backpressurePromise is undefined,
666                // set backpressurePromise to a promise resolved with undefined.
667                if backpressure_promise.borrow().is_none() {
668                    let promise = Promise::new_resolved_rooted(cx, global, ());
669                    *backpressure_promise.borrow_mut() = Some(promise.to_traced());
670                }
671
672                // Return the result of reacting to backpressurePromise with the following fulfillment steps:
673                let result_promise = Promise::new_rooted(cx, global);
674                rooted!(&in(cx) let mut fulfillment_handler = Some(TransferBackPressurePromiseReaction {
675                    port: port.clone(),
676                    backpressure_promise: backpressure_promise.clone(),
677                    chunk: Heap::boxed(chunk.get()),
678                    result_promise: result_promise.to_traced(),
679                }));
680                let handler = PromiseNativeHandler::new(
681                    cx,
682                    global,
683                    fulfillment_handler.take().map(|h| Box::new(h) as Box<_>),
684                    None,
685                );
686                let mut realm = enter_auto_realm(cx, global);
687                let realm = &mut realm.current_realm();
688                backpressure_promise
689                    .borrow()
690                    .as_ref()
691                    .expect("Promise must be some by now.")
692                    .append_native_handler(realm, &handler);
693                result_promise
694            },
695            UnderlyingSinkType::Transform(stream, _) => {
696                // Return ! TransformStreamDefaultSinkWriteAlgorithm(stream, chunk).
697                stream
698                    .transform_stream_default_sink_write_algorithm(cx, global, chunk)
699                    .expect("Transform stream default sink write algorithm should not fail.")
700            },
701        }
702    }
703
704    /// <https://streams.spec.whatwg.org/#writablestreamdefaultcontroller-closealgorithm>
705    fn call_close_algorithm(&self, cx: &mut JSContext, global: &GlobalScope) -> RootedPromise {
706        match &self.underlying_sink_type {
707            UnderlyingSinkType::Js {
708                abort: _,
709                start: _,
710                close,
711                write: _,
712            } => {
713                rooted!(&in(cx) let mut this_object = ptr::null_mut::<JSObject>());
714                this_object.set(self.underlying_sink_obj.get());
715                let algo = close.borrow().clone();
716                let result = if let Some(algo) = algo {
717                    algo.Call_(cx, &this_object.handle(), ExceptionHandling::Rethrow)
718                } else {
719                    Ok(Promise::new_resolved_rooted(cx, global, ()))
720                };
721                result.unwrap_or_else(|e| {
722                    let promise = Promise::new_rooted(cx, global);
723                    promise.reject_error(cx, e);
724                    promise
725                })
726            },
727            UnderlyingSinkType::Transfer { port, .. } => {
728                // The steps from the `closeAlgorithm` at
729                // <https://streams.spec.whatwg.org/#abstract-opdef-setupcrossrealmtransformwritable>
730
731                // Perform ! PackAndPostMessage(port, "close", undefined).
732                rooted!(&in(cx) let mut value = UndefinedValue());
733                port.pack_and_post_message(cx, "close", value.handle())
734                    .expect("Sending close should not fail.");
735
736                // Disentangle port.
737                global.disentangle_port(cx, port);
738
739                // Return a promise resolved with undefined.
740                Promise::new_resolved_rooted(cx, global, ())
741            },
742            UnderlyingSinkType::Transform(stream, _) => {
743                // Return ! TransformStreamDefaultSinkCloseAlgorithm(stream).
744                stream
745                    .transform_stream_default_sink_close_algorithm(cx, global)
746                    .expect("Transform stream default sink close algorithm should not fail.")
747            },
748        }
749    }
750
751    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-process-close>
752    pub(crate) fn process_close(&self, cx: &mut JSContext, global: &GlobalScope) {
753        // Let stream be controller.[[stream]].
754        let Some(stream) = self.stream.get() else {
755            unreachable!("Controller should have a stream");
756        };
757
758        // Perform ! WritableStreamMarkCloseRequestInFlight(stream).
759        stream.mark_close_request_in_flight();
760
761        // Perform ! DequeueValue(controller).
762        self.queue.dequeue_value(cx, None);
763
764        // Assert: controller.[[queue]] is empty.
765        assert!(self.queue.is_empty());
766
767        // Let sinkClosePromise be the result of performing controller.[[closeAlgorithm]].
768        let sink_close_promise = self.call_close_algorithm(cx, global);
769
770        // Perform ! WritableStreamDefaultControllerClearAlgorithms(controller).
771        self.clear_algorithms();
772
773        // Upon fulfillment of sinkClosePromise,
774        rooted!(&in(cx) let mut fulfillment_handler = Some(CloseAlgorithmFulfillmentHandler {
775            stream: Dom::from_ref(&stream),
776        }));
777
778        // Upon rejection of sinkClosePromise with reason reason,
779        rooted!(&in(cx) let mut rejection_handler = Some(CloseAlgorithmRejectionHandler {
780            stream: Dom::from_ref(&stream),
781        }));
782
783        // Attach handlers to the promise.
784        let handler = PromiseNativeHandler::new(
785            cx,
786            global,
787            fulfillment_handler.take().map(|h| Box::new(h) as Box<_>),
788            rejection_handler.take().map(|h| Box::new(h) as Box<_>),
789        );
790        let mut realm = enter_auto_realm(cx, global);
791        let realm = &mut realm.current_realm();
792        sink_close_promise.append_native_handler(realm, &handler);
793    }
794
795    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-advance-queue-if-needed>
796    fn advance_queue_if_needed(&self, cx: &mut JSContext, global: &GlobalScope) {
797        // Let stream be controller.[[stream]].
798        let Some(stream) = self.stream.get() else {
799            unreachable!("Controller should have a stream");
800        };
801
802        // If controller.[[started]] is false, return.
803        if !self.started.get() {
804            return;
805        }
806
807        // If stream.[[inFlightWriteRequest]] is not undefined, return.
808        if stream.has_in_flight_write_request() {
809            return;
810        }
811
812        // Let state be stream.[[state]].
813
814        // Assert: state is not "closed" or "errored".
815        assert!(!(stream.is_errored() || stream.is_closed()));
816
817        // If state is "erroring",
818        if stream.is_erroring() {
819            // Perform ! WritableStreamFinishErroring(stream).
820            stream.finish_erroring(cx, global);
821
822            // Return.
823            return;
824        }
825
826        // Let value be ! PeekQueueValue(controller).
827        rooted!(&in(cx) let mut value = UndefinedValue());
828        let is_closed = {
829            // If controller.[[queue]] is empty, return.
830            if self.queue.is_empty() {
831                return;
832            }
833            self.queue.peek_queue_value(cx, value.handle_mut())
834        };
835
836        if is_closed {
837            // If value is the close sentinel, perform ! WritableStreamDefaultControllerProcessClose(controller).
838            self.process_close(cx, global);
839        } else {
840            // Otherwise, perform ! WritableStreamDefaultControllerProcessWrite(controller, value).
841            self.process_write(cx, value.handle(), global);
842        };
843    }
844
845    /// <https://streams.spec.whatwg.org/#ws-default-controller-private-error>
846    pub(crate) fn perform_error_steps(&self) {
847        // Perform ! ResetQueue(this).
848        self.queue.reset();
849    }
850
851    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-process-write>
852    fn process_write(&self, cx: &mut JSContext, chunk: SafeHandleValue, global: &GlobalScope) {
853        // Let stream be controller.[[stream]].
854        let Some(stream) = self.stream.get() else {
855            unreachable!("Controller should have a stream");
856        };
857
858        // Perform ! WritableStreamMarkFirstWriteRequestInFlight(stream).
859        stream.mark_first_write_request_in_flight();
860
861        // Let sinkWritePromise be the result of performing controller.[[writeAlgorithm]], passing in chunk.
862        let sink_write_promise = self.call_write_algorithm(cx, chunk, global);
863
864        // Upon fulfillment of sinkWritePromise,
865        rooted!(&in(cx) let mut fulfillment_handler = Some(WriteAlgorithmFulfillmentHandler {
866            controller: Dom::from_ref(self),
867        }));
868
869        // Upon rejection of sinkWritePromise with reason,
870        rooted!(&in(cx) let mut rejection_handler = Some(WriteAlgorithmRejectionHandler {
871            controller: Dom::from_ref(self),
872        }));
873
874        // Attach handlers to the promise.
875        let handler = PromiseNativeHandler::new(
876            cx,
877            global,
878            fulfillment_handler.take().map(|h| Box::new(h) as Box<_>),
879            rejection_handler.take().map(|h| Box::new(h) as Box<_>),
880        );
881        let mut realm = enter_auto_realm(cx, global);
882        let realm = &mut realm.current_realm();
883        sink_write_promise.append_native_handler(realm, &handler);
884    }
885
886    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-get-desired-size>
887    pub(crate) fn get_desired_size(&self) -> f64 {
888        // Return controller.[[strategyHWM]] − controller.[[queueTotalSize]].
889        let desired_size = self.strategy_hwm - self.queue.total_size.get().clamp(0.0, f64::MAX);
890        desired_size.clamp(desired_size, self.strategy_hwm)
891    }
892
893    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-get-backpressure>
894    fn get_backpressure(&self) -> bool {
895        // Let desiredSize be ! WritableStreamDefaultControllerGetDesiredSize(controller).
896        let desired_size = self.get_desired_size();
897
898        // Return true if desiredSize ≤ 0, or false otherwise.
899        desired_size == 0.0 || desired_size.is_sign_negative()
900    }
901
902    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-get-chunk-size>
903    pub(crate) fn get_chunk_size(
904        &self,
905        cx: &mut JSContext,
906        global: &GlobalScope,
907        chunk: SafeHandleValue,
908    ) -> f64 {
909        // If controller.[[strategySizeAlgorithm]] is undefined, then:
910        let Some(strategy_size) = self.strategy_size.borrow().clone() else {
911            // Assert: controller.[[stream]].[[state]] is not "writable".
912            let Some(stream) = self.stream.get() else {
913                unreachable!("Controller should have a stream");
914            };
915            assert!(!stream.is_writable());
916
917            // Return 1.
918            return 1.0;
919        };
920
921        // Let returnValue be the result of performing controller.[[strategySizeAlgorithm]],
922        // passing in chunk, and interpreting the result as a completion record.
923        let result = strategy_size.Call__(cx, chunk, ExceptionHandling::Rethrow);
924
925        match result {
926            // Let chunkSize be result.[[Value]].
927            Ok(size) => size,
928            Err(error) => {
929                // If result is an abrupt completion,
930
931                // Perform ! WritableStreamDefaultControllerErrorIfNeeded(controller, returnValue.[[Value]]).
932                // Create a rooted value for the error.
933                rooted!(&in(cx) let mut rooted_error = UndefinedValue());
934                error.to_jsval(cx, global, rooted_error.handle_mut());
935                self.error_if_needed(cx, rooted_error.handle(), global);
936
937                // Return 1.
938                1.0
939            },
940        }
941    }
942
943    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-write>
944    pub(crate) fn write(
945        &self,
946        cx: &mut JSContext,
947        global: &GlobalScope,
948        chunk: SafeHandleValue,
949        chunk_size: f64,
950    ) {
951        // Let enqueueResult be EnqueueValueWithSize(controller, chunk, chunkSize).
952        let enqueue_result = self
953            .queue
954            .enqueue_value_with_size(EnqueuedValue::Js(ValueWithSize {
955                value: Heap::boxed(chunk.get()),
956                size: chunk_size,
957            }));
958
959        // If enqueueResult is an abrupt completion,
960        if let Err(error) = enqueue_result {
961            // Perform ! WritableStreamDefaultControllerErrorIfNeeded(controller, enqueueResult.[[Value]]).
962            // Create a rooted value for the error.
963            rooted!(&in(cx) let mut rooted_error = UndefinedValue());
964            error.to_jsval(cx, global, rooted_error.handle_mut());
965            self.error_if_needed(cx, rooted_error.handle(), global);
966
967            // Return.
968            return;
969        }
970
971        // Let stream be controller.[[stream]].
972        let Some(stream) = self.stream.get() else {
973            unreachable!("Controller should have a stream");
974        };
975
976        // If ! WritableStreamCloseQueuedOrInFlight(stream) is false and stream.[[state]] is "writable",
977        if !stream.close_queued_or_in_flight() && stream.is_writable() {
978            // Let backpressure be ! WritableStreamDefaultControllerGetBackpressure(controller).
979            let backpressure = self.get_backpressure();
980
981            // Perform ! WritableStreamUpdateBackpressure(stream, backpressure).
982            stream.update_backpressure(cx, backpressure, global);
983        }
984
985        // Perform ! WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller).
986        self.advance_queue_if_needed(cx, global);
987    }
988
989    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-error-if-needed>
990    pub(crate) fn error_if_needed(
991        &self,
992        cx: &mut JSContext,
993        error: SafeHandleValue,
994        global: &GlobalScope,
995    ) {
996        // Let stream be controller.[[stream]].
997        let Some(stream) = self.stream.get() else {
998            unreachable!("Controller should have a stream");
999        };
1000
1001        // If stream.[[state]] is "writable",
1002        if stream.is_writable() {
1003            // Perform ! WritableStreamDefaultControllerError(controller, e).
1004            self.error(cx, &stream, error, global);
1005        }
1006    }
1007
1008    /// <https://streams.spec.whatwg.org/#writable-stream-default-controller-error>
1009    fn error(
1010        &self,
1011        cx: &mut JSContext,
1012        stream: &WritableStream,
1013        e: SafeHandleValue,
1014        global: &GlobalScope,
1015    ) {
1016        // Let stream be controller.[[stream]].
1017        // Done above with the argument.
1018
1019        // Assert: stream.[[state]] is "writable".
1020        assert!(stream.is_writable());
1021
1022        // Perform ! WritableStreamDefaultControllerClearAlgorithms(controller).
1023        self.clear_algorithms();
1024
1025        // Perform ! WritableStreamStartErroring(stream, error).
1026        stream.start_erroring(cx, global, e);
1027    }
1028}
1029
1030impl WritableStreamDefaultControllerMethods<crate::DomTypeHolder>
1031    for WritableStreamDefaultController
1032{
1033    /// <https://streams.spec.whatwg.org/#ws-default-controller-error>
1034    fn Error(&self, cx: &mut CurrentRealm, e: SafeHandleValue) {
1035        // Let state be this.[[stream]].[[state]].
1036        let Some(stream) = self.stream.get() else {
1037            unreachable!("Controller should have a stream");
1038        };
1039
1040        // If state is not "writable", return.
1041        if !stream.is_writable() {
1042            return;
1043        }
1044
1045        let global = GlobalScope::from_current_realm(cx);
1046
1047        // Perform ! WritableStreamDefaultControllerError(this, e).
1048        self.error(cx, &stream, e, &global);
1049    }
1050
1051    /// <https://streams.spec.whatwg.org/#ws-default-controller-signal>
1052    fn Signal(&self) -> DomRoot<AbortSignal> {
1053        // Return this.[[abortController]]’s signal.
1054        self.abort_controller.signal()
1055    }
1056}