script/dom/stream/
readablestreamdefaultcontroller.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 http://mozilla.org/MPL/2.0/. */
4
5use std::cell::{Cell, RefCell};
6use std::collections::VecDeque;
7use std::ptr;
8use std::rc::Rc;
9
10use dom_struct::dom_struct;
11use js::jsapi::{Heap, JSObject};
12use js::jsval::{JSVal, UndefinedValue};
13use js::realm::CurrentRealm;
14use js::rust::wrappers2::JS_GetPendingException;
15use js::rust::{HandleObject, HandleValue as SafeHandleValue, HandleValue, MutableHandleValue};
16use js::typedarray::Uint8;
17use script_bindings::conversions::SafeToJSValConvertible;
18
19use crate::dom::bindings::buffer_source::create_buffer_source;
20use crate::dom::bindings::callback::ExceptionHandling;
21use crate::dom::bindings::codegen::Bindings::QueuingStrategyBinding::QueuingStrategySize;
22use crate::dom::bindings::codegen::Bindings::ReadableStreamDefaultControllerBinding::ReadableStreamDefaultControllerMethods;
23use crate::dom::bindings::codegen::UnionTypes::ReadableStreamDefaultControllerOrReadableByteStreamController as Controller;
24use crate::dom::bindings::error::{Error, ErrorToJsval, Fallible, throw_dom_exception};
25use crate::dom::bindings::reflector::{DomGlobal, Reflector, reflect_dom_object};
26use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
27use crate::dom::bindings::trace::RootedTraceableBox;
28use crate::dom::globalscope::GlobalScope;
29use crate::dom::promise::Promise;
30use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
31use crate::dom::stream::readablestream::ReadableStream;
32use crate::dom::stream::readablestreamdefaultreader::ReadRequest;
33use crate::dom::stream::underlyingsourcecontainer::{
34    UnderlyingSourceContainer, UnderlyingSourceType,
35};
36use crate::realms::{InRealm, enter_realm};
37use crate::script_runtime::{CanGc, JSContext as SafeJSContext};
38
39/// The fulfillment handler for
40/// <https://streams.spec.whatwg.org/#readable-stream-default-controller-call-pull-if-needed>
41#[derive(Clone, JSTraceable, MallocSizeOf)]
42#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
43struct PullAlgorithmFulfillmentHandler {
44    controller: Dom<ReadableStreamDefaultController>,
45}
46
47impl Callback for PullAlgorithmFulfillmentHandler {
48    /// Continuation of <https://streams.spec.whatwg.org/#readable-stream-default-controller-call-pull-if-needed>
49    /// Upon fulfillment of pullPromise
50    fn callback(&self, cx: &mut CurrentRealm, _v: HandleValue) {
51        let can_gc = CanGc::from_cx(cx);
52        // Set controller.[[pulling]] to false.
53        self.controller.pulling.set(false);
54
55        // If controller.[[pullAgain]] is true,
56        if self.controller.pull_again.get() {
57            // Set controller.[[pullAgain]] to false.
58            self.controller.pull_again.set(false);
59
60            // Perform ! ReadableStreamDefaultControllerCallPullIfNeeded(controller).
61            self.controller.call_pull_if_needed(can_gc);
62        }
63    }
64}
65
66/// The rejection handler for
67/// <https://streams.spec.whatwg.org/#readable-stream-default-controller-call-pull-if-needed>
68#[derive(Clone, JSTraceable, MallocSizeOf)]
69#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
70struct PullAlgorithmRejectionHandler {
71    controller: Dom<ReadableStreamDefaultController>,
72}
73
74impl Callback for PullAlgorithmRejectionHandler {
75    /// Continuation of <https://streams.spec.whatwg.org/#readable-stream-default-controller-call-pull-if-needed>
76    /// Upon rejection of pullPromise with reason e.
77    fn callback(&self, cx: &mut CurrentRealm, v: HandleValue) {
78        let can_gc = CanGc::from_cx(cx);
79        // Perform ! ReadableStreamDefaultControllerError(controller, e).
80        self.controller.error(v, can_gc);
81    }
82}
83
84/// The fulfillment handler for
85/// <https://streams.spec.whatwg.org/#dom-underlyingsource-start>
86#[derive(Clone, JSTraceable, MallocSizeOf)]
87#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
88struct StartAlgorithmFulfillmentHandler {
89    controller: Dom<ReadableStreamDefaultController>,
90}
91
92impl Callback for StartAlgorithmFulfillmentHandler {
93    /// Continuation of <https://streams.spec.whatwg.org/#set-up-readable-stream-default-controller>
94    /// Upon fulfillment of startPromise,
95    fn callback(&self, cx: &mut CurrentRealm, _v: HandleValue) {
96        let can_gc = CanGc::from_cx(cx);
97        // Set controller.[[started]] to true.
98        self.controller.started.set(true);
99
100        // Perform ! ReadableStreamDefaultControllerCallPullIfNeeded(controller).
101        self.controller.call_pull_if_needed(can_gc);
102    }
103}
104
105/// The rejection handler for
106/// <https://streams.spec.whatwg.org/#dom-underlyingsource-start>
107#[derive(Clone, JSTraceable, MallocSizeOf)]
108#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
109struct StartAlgorithmRejectionHandler {
110    controller: Dom<ReadableStreamDefaultController>,
111}
112
113impl Callback for StartAlgorithmRejectionHandler {
114    /// Continuation of <https://streams.spec.whatwg.org/#set-up-readable-stream-default-controller>
115    /// Upon rejection of startPromise with reason r,
116    fn callback(&self, cx: &mut CurrentRealm, v: HandleValue) {
117        let can_gc = CanGc::from_cx(cx);
118        // Perform ! ReadableStreamDefaultControllerError(controller, r).
119        self.controller.error(v, can_gc);
120    }
121}
122
123/// <https://streams.spec.whatwg.org/#value-with-size>
124#[derive(Debug, JSTraceable, MallocSizeOf, PartialEq)]
125#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
126pub(crate) struct ValueWithSize {
127    /// <https://streams.spec.whatwg.org/#value-with-size-value>
128    #[ignore_malloc_size_of = "Heap is measured by mozjs"]
129    pub(crate) value: Box<Heap<JSVal>>,
130    /// <https://streams.spec.whatwg.org/#value-with-size-size>
131    pub(crate) size: f64,
132}
133
134/// <https://streams.spec.whatwg.org/#value-with-size>
135#[derive(Debug, JSTraceable, MallocSizeOf, PartialEq)]
136#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
137pub(crate) enum EnqueuedValue {
138    /// A value enqueued from Rust.
139    Native(Box<[u8]>),
140    /// A Js value.
141    Js(ValueWithSize),
142    /// <https://streams.spec.whatwg.org/#close-sentinel>
143    CloseSentinel,
144}
145
146impl EnqueuedValue {
147    fn size(&self) -> f64 {
148        match self {
149            EnqueuedValue::Native(v) => v.len() as f64,
150            EnqueuedValue::Js(v) => v.size,
151            // The size of the sentinel is zero,
152            // as per <https://streams.spec.whatwg.org/#ref-for-close-sentinel%E2%91%A0>
153            EnqueuedValue::CloseSentinel => 0.,
154        }
155    }
156
157    fn to_jsval(&self, cx: SafeJSContext, rval: MutableHandleValue, can_gc: CanGc) {
158        match self {
159            EnqueuedValue::Native(chunk) => {
160                rooted!(in(*cx) let mut array_buffer_ptr = ptr::null_mut::<JSObject>());
161                create_buffer_source::<Uint8>(cx, chunk, array_buffer_ptr.handle_mut(), can_gc)
162                    .expect("failed to create buffer source for native chunk.");
163                array_buffer_ptr.safe_to_jsval(cx, rval, can_gc);
164            },
165            EnqueuedValue::Js(value_with_size) => {
166                value_with_size.value.safe_to_jsval(cx, rval, can_gc)
167            },
168            EnqueuedValue::CloseSentinel => {
169                unreachable!("The close sentinel is never made available as a js val.")
170            },
171        }
172    }
173}
174
175/// <https://streams.spec.whatwg.org/#is-non-negative-number>
176fn is_non_negative_number(value: &EnqueuedValue) -> bool {
177    let value_with_size = match value {
178        EnqueuedValue::Native(_) => return true,
179        EnqueuedValue::Js(value_with_size) => value_with_size,
180        EnqueuedValue::CloseSentinel => return true,
181    };
182
183    // If v is not a Number, return false.
184    // Checked as part of the WebIDL.
185
186    // If v is NaN, return false.
187    if value_with_size.size.is_nan() {
188        return false;
189    }
190
191    // If v < 0, return false.
192    if value_with_size.size.is_sign_negative() {
193        return false;
194    }
195
196    true
197}
198
199/// <https://streams.spec.whatwg.org/#queue-with-sizes>
200#[derive(Default, JSTraceable, MallocSizeOf)]
201#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
202pub(crate) struct QueueWithSizes {
203    queue: RefCell<VecDeque<EnqueuedValue>>,
204    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-queuetotalsize>
205    pub(crate) total_size: Cell<f64>,
206}
207
208impl QueueWithSizes {
209    /// <https://streams.spec.whatwg.org/#dequeue-value>
210    /// A none `rval` means we're dequeing the close sentinel,
211    /// which should never be made available to script.
212    pub(crate) fn dequeue_value(
213        &self,
214        cx: SafeJSContext,
215        rval: Option<MutableHandleValue>,
216        can_gc: CanGc,
217    ) {
218        {
219            let queue = self.queue.borrow();
220            let Some(value) = queue.front() else {
221                unreachable!("Buffer cannot be empty when dequeue value is called into.");
222            };
223            self.total_size.set(self.total_size.get() - value.size());
224            if let Some(rval) = rval {
225                value.to_jsval(cx, rval, can_gc);
226            } else {
227                assert_eq!(value, &EnqueuedValue::CloseSentinel);
228            }
229        }
230        self.queue.borrow_mut().pop_front();
231    }
232
233    /// <https://streams.spec.whatwg.org/#enqueue-value-with-size>
234    #[cfg_attr(crown, expect(crown::unrooted_must_root))]
235    pub(crate) fn enqueue_value_with_size(&self, value: EnqueuedValue) -> Result<(), Error> {
236        // If ! IsNonNegativeNumber(size) is false, throw a RangeError exception.
237        if !is_non_negative_number(&value) {
238            return Err(Error::Range(
239                c"The size of the enqueued chunk is not a non-negative number.".to_owned(),
240            ));
241        }
242
243        // If size is +∞, throw a RangeError exception.
244        if value.size().is_infinite() {
245            return Err(Error::Range(
246                c"The size of the enqueued chunk is infinite.".to_owned(),
247            ));
248        }
249
250        self.total_size.set(self.total_size.get() + value.size());
251        self.queue.borrow_mut().push_back(value);
252
253        Ok(())
254    }
255
256    pub(crate) fn is_empty(&self) -> bool {
257        self.queue.borrow().is_empty()
258    }
259
260    /// <https://streams.spec.whatwg.org/#peek-queue-value>
261    /// Returns whether value is the close sentinel.
262    pub(crate) fn peek_queue_value(
263        &self,
264        cx: SafeJSContext,
265        rval: MutableHandleValue,
266        can_gc: CanGc,
267    ) -> bool {
268        // Assert: container has [[queue]] and [[queueTotalSize]] internal slots.
269        // Done with the QueueWithSizes type.
270
271        // Assert: container.[[queue]] is not empty.
272        assert!(!self.is_empty());
273
274        // Let valueWithSize be container.[[queue]][0].
275        let queue = self.queue.borrow();
276        let value_with_size = queue.front().expect("Queue is not empty.");
277        if let EnqueuedValue::CloseSentinel = value_with_size {
278            return true;
279        }
280
281        // Return valueWithSize’s value.
282        value_with_size.to_jsval(cx, rval, can_gc);
283        false
284    }
285
286    /// Only used with native sources.
287    fn get_in_memory_bytes(&self) -> Option<Vec<u8>> {
288        self.queue
289            .borrow()
290            .iter()
291            .try_fold(Vec::new(), |mut acc, value| match value {
292                EnqueuedValue::Native(chunk) => {
293                    acc.extend(chunk.iter().copied());
294                    Some(acc)
295                },
296                _ => {
297                    warn!("get_in_memory_bytes called on a controller with non-native source.");
298                    None
299                },
300            })
301    }
302
303    /// <https://streams.spec.whatwg.org/#reset-queue>
304    pub(crate) fn reset(&self) {
305        self.queue.borrow_mut().clear();
306        self.total_size.set(Default::default());
307    }
308}
309
310/// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller>
311#[dom_struct]
312pub(crate) struct ReadableStreamDefaultController {
313    reflector_: Reflector,
314
315    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-queue>
316    queue: QueueWithSizes,
317
318    /// A mutable reference to the underlying source is used to implement these two
319    /// internal slots:
320    ///
321    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-pullalgorithm>
322    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-cancelalgorithm>
323    underlying_source: MutNullableDom<UnderlyingSourceContainer>,
324
325    stream: MutNullableDom<ReadableStream>,
326
327    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-strategyhwm>
328    strategy_hwm: f64,
329
330    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-strategysizealgorithm>
331    #[ignore_malloc_size_of = "mozjs"]
332    strategy_size: RefCell<Option<Rc<QueuingStrategySize>>>,
333
334    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-closerequested>
335    close_requested: Cell<bool>,
336
337    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-started>
338    started: Cell<bool>,
339
340    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-pulling>
341    pulling: Cell<bool>,
342
343    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-pullagain>
344    pull_again: Cell<bool>,
345}
346
347impl ReadableStreamDefaultController {
348    #[cfg_attr(crown, expect(crown::unrooted_must_root))]
349    fn new_inherited(
350        global: &GlobalScope,
351        underlying_source_type: UnderlyingSourceType,
352        strategy_hwm: f64,
353        strategy_size: Rc<QueuingStrategySize>,
354        can_gc: CanGc,
355    ) -> ReadableStreamDefaultController {
356        ReadableStreamDefaultController {
357            reflector_: Reflector::new(),
358            queue: Default::default(),
359            stream: MutNullableDom::new(None),
360            underlying_source: MutNullableDom::new(Some(&*UnderlyingSourceContainer::new(
361                global,
362                underlying_source_type,
363                can_gc,
364            ))),
365            strategy_hwm,
366            strategy_size: RefCell::new(Some(strategy_size)),
367            close_requested: Default::default(),
368            started: Default::default(),
369            pulling: Default::default(),
370            pull_again: Default::default(),
371        }
372    }
373
374    #[cfg_attr(crown, expect(crown::unrooted_must_root))]
375    pub(crate) fn new(
376        global: &GlobalScope,
377        underlying_source: UnderlyingSourceType,
378        strategy_hwm: f64,
379        strategy_size: Rc<QueuingStrategySize>,
380        can_gc: CanGc,
381    ) -> DomRoot<ReadableStreamDefaultController> {
382        reflect_dom_object(
383            Box::new(ReadableStreamDefaultController::new_inherited(
384                global,
385                underlying_source,
386                strategy_hwm,
387                strategy_size,
388                can_gc,
389            )),
390            global,
391            can_gc,
392        )
393    }
394
395    /// <https://streams.spec.whatwg.org/#set-up-readable-stream-default-controller>
396    pub(crate) fn setup(
397        &self,
398        stream: DomRoot<ReadableStream>,
399        can_gc: CanGc,
400    ) -> Result<(), Error> {
401        // Assert: stream.[[controller]] is undefined
402        stream.assert_no_controller();
403
404        // Set controller.[[stream]] to stream.
405        self.stream.set(Some(&stream));
406
407        let global = &*self.global();
408        let rooted_default_controller = DomRoot::from_ref(self);
409
410        // Perform ! ResetQueue(controller).
411        // Set controller.[[started]], controller.[[closeRequested]],
412        // controller.[[pullAgain]], and controller.[[pulling]] to false.
413        // Set controller.[[strategySizeAlgorithm]] to sizeAlgorithm
414        // and controller.[[strategyHWM]] to highWaterMark.
415        // Set controller.[[strategySizeAlgorithm]] to sizeAlgorithm
416        // and controller.[[strategyHWM]] to highWaterMark.
417        // Set controller.[[cancelAlgorithm]] to cancelAlgorithm.
418
419        // Note: the above steps are done in `new`.
420
421        // Set stream.[[controller]] to controller.
422        stream.set_default_controller(&rooted_default_controller);
423
424        if let Some(underlying_source) = rooted_default_controller.underlying_source.get() {
425            // Let startResult be the result of performing startAlgorithm. (This might throw an exception.)
426            let start_result = underlying_source
427                .call_start_algorithm(
428                    Controller::ReadableStreamDefaultController(rooted_default_controller.clone()),
429                    can_gc,
430                )
431                .unwrap_or_else(|| {
432                    let promise = Promise::new(global, can_gc);
433                    promise.resolve_native(&(), can_gc);
434                    Ok(promise)
435                });
436
437            // Let startPromise be a promise resolved with startResult.
438            let start_promise = start_result?;
439
440            // Upon fulfillment of startPromise, Upon rejection of startPromise with reason r,
441            let handler = PromiseNativeHandler::new(
442                global,
443                Some(Box::new(StartAlgorithmFulfillmentHandler {
444                    controller: Dom::from_ref(&rooted_default_controller),
445                })),
446                Some(Box::new(StartAlgorithmRejectionHandler {
447                    controller: Dom::from_ref(&rooted_default_controller),
448                })),
449                can_gc,
450            );
451            let realm = enter_realm(global);
452            let comp = InRealm::Entered(&realm);
453            start_promise.append_native_handler(&handler, comp, can_gc);
454        };
455
456        Ok(())
457    }
458
459    /// Setting the JS object after the heap has settled down.
460    pub(crate) fn set_underlying_source_this_object(&self, this_object: HandleObject) {
461        if let Some(underlying_source) = self.underlying_source.get() {
462            underlying_source.set_underlying_source_this_object(this_object);
463        }
464    }
465
466    /// <https://streams.spec.whatwg.org/#dequeue-value>
467    fn dequeue_value(&self, cx: SafeJSContext, rval: MutableHandleValue, can_gc: CanGc) {
468        self.queue.dequeue_value(cx, Some(rval), can_gc);
469    }
470
471    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-should-call-pull>
472    fn should_call_pull(&self) -> bool {
473        // Let stream be controller.[[stream]].
474        // Note: the spec does not assert that stream is not undefined here,
475        // so we return false if it is.
476        let Some(stream) = self.stream.get() else {
477            debug!("`should_call_pull` called on a controller without a stream.");
478            return false;
479        };
480
481        // If ! ReadableStreamDefaultControllerCanCloseOrEnqueue(controller) is false, return.
482        if !self.can_close_or_enqueue() {
483            return false;
484        }
485
486        // If controller.[[started]] is false, return false.
487        if !self.started.get() {
488            return false;
489        }
490
491        // If ! IsReadableStreamLocked(stream) is true
492        // and ! ReadableStreamGetNumReadRequests(stream) > 0, return true.
493        if stream.is_locked() && stream.get_num_read_requests() > 0 {
494            return true;
495        }
496
497        // Let desiredSize be ! ReadableStreamDefaultControllerGetDesiredSize(controller).
498        // Assert: desiredSize is not null.
499        let desired_size = self.get_desired_size().expect("desiredSize is not null.");
500
501        if desired_size > 0. {
502            return true;
503        }
504
505        false
506    }
507
508    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-call-pull-if-needed>
509    fn call_pull_if_needed(&self, can_gc: CanGc) {
510        // Let shouldPull be ! ReadableStreamDefaultControllerShouldCallPull(controller).
511        // If shouldPull is false, return.
512        if !self.should_call_pull() {
513            return;
514        }
515
516        // If controller.[[pulling]] is true,
517        if self.pulling.get() {
518            // Set controller.[[pullAgain]] to true.
519            self.pull_again.set(true);
520
521            return;
522        }
523
524        // Set controller.[[pulling]] to true.
525        self.pulling.set(true);
526
527        // Let pullPromise be the result of performing controller.[[pullAlgorithm]].
528        // Continues into the resolve and reject handling of the native handler.
529        let global = self.global();
530        let rooted_default_controller = DomRoot::from_ref(self);
531        let controller =
532            Controller::ReadableStreamDefaultController(rooted_default_controller.clone());
533
534        let Some(underlying_source) = self.underlying_source.get() else {
535            return;
536        };
537        let handler = PromiseNativeHandler::new(
538            &global,
539            Some(Box::new(PullAlgorithmFulfillmentHandler {
540                controller: Dom::from_ref(&rooted_default_controller),
541            })),
542            Some(Box::new(PullAlgorithmRejectionHandler {
543                controller: Dom::from_ref(&rooted_default_controller),
544            })),
545            can_gc,
546        );
547
548        let realm = enter_realm(&*global);
549        let comp = InRealm::Entered(&realm);
550        let result = underlying_source
551            .call_pull_algorithm(controller, can_gc)
552            .unwrap_or_else(|| {
553                let promise = Promise::new(&global, can_gc);
554                promise.resolve_native(&(), can_gc);
555                Ok(promise)
556            });
557        let promise = result.unwrap_or_else(|error| {
558            let cx = GlobalScope::get_cx();
559            rooted!(in(*cx) let mut rval = UndefinedValue());
560            // TODO: check if `self.global()` is the right globalscope.
561            error.to_jsval(cx, &self.global(), rval.handle_mut(), can_gc);
562            let promise = Promise::new(&global, can_gc);
563            promise.reject_native(&rval.handle(), can_gc);
564            promise
565        });
566        promise.append_native_handler(&handler, comp, can_gc);
567    }
568
569    /// <https://streams.spec.whatwg.org/#rs-default-controller-private-cancel>
570    pub(crate) fn perform_cancel_steps(
571        &self,
572        cx: &mut js::context::JSContext,
573        global: &GlobalScope,
574        reason: SafeHandleValue,
575    ) -> Rc<Promise> {
576        // Perform ! ResetQueue(this).
577        self.queue.reset();
578
579        let underlying_source = self
580            .underlying_source
581            .get()
582            .expect("Controller should have a source when the cancel steps are called into.");
583        // Let result be the result of performing this.[[cancelAlgorithm]], passing reason.
584        let result = underlying_source
585            .call_cancel_algorithm(cx, global, reason)
586            .unwrap_or_else(|| {
587                let promise = Promise::new2(cx, global);
588                promise.resolve_native(&(), CanGc::from_cx(cx));
589                Ok(promise)
590            });
591        let promise = result.unwrap_or_else(|error| {
592            rooted!(&in(cx) let mut rval = UndefinedValue());
593
594            error.to_jsval(cx.into(), global, rval.handle_mut(), CanGc::from_cx(cx));
595            let promise = Promise::new2(cx, global);
596            promise.reject_native(&rval.handle(), CanGc::from_cx(cx));
597            promise
598        });
599
600        // Perform ! ReadableStreamDefaultControllerClearAlgorithms(this).
601        self.clear_algorithms();
602
603        // Return result(the promise).
604        promise
605    }
606
607    /// <https://streams.spec.whatwg.org/#rs-default-controller-private-pull>
608    pub(crate) fn perform_pull_steps(&self, read_request: &ReadRequest, can_gc: CanGc) {
609        // Let stream be this.[[stream]].
610        // Note: the spec does not assert that there is a stream.
611        let Some(stream) = self.stream.get() else {
612            return;
613        };
614
615        // if queue contains bytes, perform chunk steps.
616        if !self.queue.is_empty() {
617            let cx = GlobalScope::get_cx();
618            rooted!(in(*cx) let mut rval = UndefinedValue());
619            let result = RootedTraceableBox::new(Heap::default());
620            self.dequeue_value(cx, rval.handle_mut(), can_gc);
621            result.set(*rval);
622
623            // If this.[[closeRequested]] is true and this.[[queue]] is empty
624            if self.close_requested.get() && self.queue.is_empty() {
625                // Perform ! ReadableStreamDefaultControllerClearAlgorithms(controller).
626                self.clear_algorithms();
627
628                // Perform ! ReadableStreamClose(stream).
629                stream.close(can_gc);
630            } else {
631                // Otherwise, perform ! ReadableStreamDefaultControllerCallPullIfNeeded(this).
632                self.call_pull_if_needed(can_gc);
633            }
634            // Perform readRequest’s chunk steps, given chunk.
635            read_request.chunk_steps(result, &self.global(), can_gc);
636        } else {
637            // Perform ! ReadableStreamAddReadRequest(stream, readRequest).
638            stream.add_read_request(read_request);
639
640            // Perform ! ReadableStreamDefaultControllerCallPullIfNeeded(this).
641            self.call_pull_if_needed(can_gc);
642        }
643    }
644
645    /// <https://streams.spec.whatwg.org/#ref-for-abstract-opdef-readablestreamcontroller-releasesteps>
646    pub(crate) fn perform_release_steps(&self) -> Fallible<()> {
647        // step 1 - Return.
648        Ok(())
649    }
650
651    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-enqueue>
652    #[expect(unsafe_code)]
653    pub(crate) fn enqueue(
654        &self,
655        cx: &mut js::context::JSContext,
656        chunk: SafeHandleValue,
657    ) -> Result<(), Error> {
658        // If ! ReadableStreamDefaultControllerCanCloseOrEnqueue(controller) is false, return.
659        if !self.can_close_or_enqueue() {
660            return Ok(());
661        }
662
663        let stream = self
664            .stream
665            .get()
666            .expect("Controller must have a stream when a chunk is enqueued.");
667
668        // If ! IsReadableStreamLocked(stream) is true
669        // and ! ReadableStreamGetNumReadRequests(stream) > 0,
670        // perform ! ReadableStreamFulfillReadRequest(stream, chunk, false).
671        if stream.is_locked() && stream.get_num_read_requests() > 0 {
672            stream.fulfill_read_request(chunk, false, CanGc::from_cx(cx));
673        } else {
674            // Otherwise,
675            // Let result be the result of performing controller.[[strategySizeAlgorithm]],
676            // passing in chunk, and interpreting the result as a completion record.
677            // Note: the clone is necessary to prevent potential re-borrow panics.
678            let strategy_size = {
679                let reference = self.strategy_size.borrow();
680                reference.clone()
681            };
682            let size = if let Some(strategy_size) = strategy_size {
683                // Note: the Rethrow exception handling is necessary,
684                // otherwise returning JSFailed will panic because no exception is pending.
685                let result =
686                    strategy_size.Call__(chunk, ExceptionHandling::Rethrow, CanGc::from_cx(cx));
687                match result {
688                    // Let chunkSize be result.[[Value]].
689                    Ok(size) => size,
690                    Err(error) => {
691                        // If result is an abrupt completion,
692                        rooted!(&in(cx) let mut rval = UndefinedValue());
693                        unsafe { assert!(JS_GetPendingException(cx, rval.handle_mut())) };
694
695                        // Perform ! ReadableStreamDefaultControllerError(controller, result.[[Value]]).
696                        self.error(rval.handle(), CanGc::from_cx(cx));
697
698                        // Return result.
699                        // Note: we need to return a type error, because no exception is pending.
700                        return Err(error);
701                    },
702                }
703            } else {
704                0.
705            };
706
707            {
708                // Let enqueueResult be EnqueueValueWithSize(controller, chunk, chunkSize).
709                let res = self
710                    .queue
711                    .enqueue_value_with_size(EnqueuedValue::Js(ValueWithSize {
712                        value: Heap::boxed(chunk.get()),
713                        size,
714                    }));
715                if let Err(error) = res {
716                    // If enqueueResult is an abrupt completion,
717
718                    // First, throw the exception.
719                    // Note: this must be done manually here,
720                    // because `enqueue_value_with_size` does not call into JS.
721                    throw_dom_exception(cx.into(), &self.global(), error, CanGc::from_cx(cx));
722
723                    // Then, get a handle to the JS val for the exception,
724                    // and use that to error the stream.
725                    rooted!(&in(cx) let mut rval = UndefinedValue());
726                    unsafe { assert!(JS_GetPendingException(cx, rval.handle_mut())) };
727
728                    // Perform ! ReadableStreamDefaultControllerError(controller, enqueueResult.[[Value]]).
729                    self.error(rval.handle(), CanGc::from_cx(cx));
730
731                    // Return enqueueResult.
732                    // Note: because we threw the exception above,
733                    // there is a pending exception and we can return JSFailed.
734                    return Err(Error::JSFailed);
735                }
736            }
737        }
738
739        // Perform ! ReadableStreamDefaultControllerCallPullIfNeeded(controller).
740        self.call_pull_if_needed(CanGc::from_cx(cx));
741
742        Ok(())
743    }
744
745    /// Native call to
746    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-enqueue>
747    pub(crate) fn enqueue_native(&self, chunk: Vec<u8>, can_gc: CanGc) {
748        let stream = self
749            .stream
750            .get()
751            .expect("Controller must have a stream when a chunk is enqueued.");
752        if stream.is_locked() && stream.get_num_read_requests() > 0 {
753            let cx = GlobalScope::get_cx();
754            rooted!(in(*cx) let mut rval = UndefinedValue());
755            EnqueuedValue::Native(chunk.into_boxed_slice()).to_jsval(cx, rval.handle_mut(), can_gc);
756            stream.fulfill_read_request(rval.handle(), false, can_gc);
757        } else {
758            self.queue
759                .enqueue_value_with_size(EnqueuedValue::Native(chunk.into_boxed_slice()))
760                .expect("Enqueuing a chunk from Rust should not fail.");
761        }
762    }
763
764    /// Does the stream have all data in memory?
765    pub(crate) fn in_memory(&self) -> bool {
766        let Some(underlying_source) = self.underlying_source.get() else {
767            return false;
768        };
769        underlying_source.in_memory()
770    }
771
772    /// Return bytes synchronously if the stream has all data in memory.
773    pub(crate) fn get_in_memory_bytes(&self) -> Option<Vec<u8>> {
774        let underlying_source = self.underlying_source.get()?;
775        if underlying_source.in_memory() {
776            return self.queue.get_in_memory_bytes();
777        }
778        None
779    }
780
781    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-clear-algorithms>
782    fn clear_algorithms(&self) {
783        // Set controller.[[pullAlgorithm]] to undefined.
784        // Set controller.[[cancelAlgorithm]] to undefined.
785        self.underlying_source.set(None);
786
787        // Set controller.[[strategySizeAlgorithm]] to undefined.
788        *self.strategy_size.borrow_mut() = None;
789    }
790
791    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-close>
792    pub(crate) fn close(&self, can_gc: CanGc) {
793        // If ! ReadableStreamDefaultControllerCanCloseOrEnqueue(controller) is false, return.
794        if !self.can_close_or_enqueue() {
795            return;
796        }
797
798        let Some(stream) = self.stream.get() else {
799            return;
800        };
801
802        // Set controller.[[closeRequested]] to true.
803        self.close_requested.set(true);
804
805        if self.queue.is_empty() {
806            // Perform ! ReadableStreamDefaultControllerClearAlgorithms(controller).
807            self.clear_algorithms();
808
809            // Perform ! ReadableStreamClose(stream).
810            stream.close(can_gc);
811        }
812    }
813
814    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-get-desired-size>
815    pub(crate) fn get_desired_size(&self) -> Option<f64> {
816        let stream = self.stream.get()?;
817
818        // If state is "errored", return null.
819        if stream.is_errored() {
820            return None;
821        }
822
823        // If state is "closed", return 0.
824        if stream.is_closed() {
825            return Some(0.0);
826        }
827
828        // Return controller.[[strategyHWM]] − controller.[[queueTotalSize]].
829        let desired_size = self.strategy_hwm - self.queue.total_size.get().clamp(0.0, f64::MAX);
830        Some(desired_size.clamp(desired_size, self.strategy_hwm))
831    }
832
833    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-can-close-or-enqueue>
834    pub(crate) fn can_close_or_enqueue(&self) -> bool {
835        let Some(stream) = self.stream.get() else {
836            return false;
837        };
838
839        // If controller.[[closeRequested]] is false and state is "readable", return true.
840        if !self.close_requested.get() && stream.is_readable() {
841            return true;
842        }
843
844        // Otherwise, return false.
845        false
846    }
847
848    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-error>
849    pub(crate) fn error(&self, e: SafeHandleValue, can_gc: CanGc) {
850        let Some(stream) = self.stream.get() else {
851            return;
852        };
853
854        // If stream.[[state]] is not "readable", return.
855        if !stream.is_readable() {
856            return;
857        }
858
859        // Perform ! ResetQueue(controller).
860        self.queue.reset();
861
862        // Perform ! ReadableStreamDefaultControllerClearAlgorithms(controller).
863        self.clear_algorithms();
864
865        stream.error(e, can_gc);
866    }
867
868    /// <https://streams.spec.whatwg.org/#rs-default-controller-has-backpressure>
869    pub(crate) fn has_backpressure(&self) -> bool {
870        // If ! ReadableStreamDefaultControllerShouldCallPull(controller) is true, return false.
871        // Otherwise, return true.
872        !self.should_call_pull()
873    }
874}
875
876impl ReadableStreamDefaultControllerMethods<crate::DomTypeHolder>
877    for ReadableStreamDefaultController
878{
879    /// <https://streams.spec.whatwg.org/#rs-default-controller-desired-size>
880    fn GetDesiredSize(&self) -> Option<f64> {
881        self.get_desired_size()
882    }
883
884    /// <https://streams.spec.whatwg.org/#rs-default-controller-close>
885    fn Close(&self, can_gc: CanGc) -> Fallible<()> {
886        if !self.can_close_or_enqueue() {
887            // If ! ReadableStreamDefaultControllerCanCloseOrEnqueue(this) is false,
888            // throw a TypeError exception.
889            return Err(Error::Type(c"Stream cannot be closed.".to_owned()));
890        }
891
892        // Perform ! ReadableStreamDefaultControllerClose(this).
893        self.close(can_gc);
894
895        Ok(())
896    }
897
898    /// <https://streams.spec.whatwg.org/#rs-default-controller-enqueue>
899    fn Enqueue(&self, cx: &mut js::context::JSContext, chunk: SafeHandleValue) -> Fallible<()> {
900        // If ! ReadableStreamDefaultControllerCanCloseOrEnqueue(this) is false, throw a TypeError exception.
901        if !self.can_close_or_enqueue() {
902            return Err(Error::Type(c"Stream cannot be enqueued to.".to_owned()));
903        }
904
905        // Perform ? ReadableStreamDefaultControllerEnqueue(this, chunk).
906        self.enqueue(cx, chunk)
907    }
908
909    /// <https://streams.spec.whatwg.org/#rs-default-controller-error>
910    fn Error(&self, cx: &mut js::context::JSContext, e: SafeHandleValue) -> Fallible<()> {
911        self.error(e, CanGc::from_cx(cx));
912        Ok(())
913    }
914}
script/dom/stream/readablestreamdefaultcontroller.rs

script/dom/stream/
readablestreamdefaultcontroller.rs
