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

script/dom/stream/
readablestreamdefaultcontroller.rs
