script/dom/stream/
defaultteereadrequest.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;
6use std::rc::Rc;
7
8use dom_struct::dom_struct;
9use js::context::JSContext;
10use js::jsapi::Heap;
11use js::jsval::{JSVal, UndefinedValue};
12use js::rust::HandleValue as SafeHandleValue;
13use script_bindings::reflector::{Reflector, reflect_dom_object};
14
15use crate::dom::bindings::error::ErrorToJsval;
16use crate::dom::bindings::reflector::DomGlobal;
17use crate::dom::bindings::root::{Dom, DomRoot};
18use crate::dom::bindings::structuredclone;
19use crate::dom::bindings::trace::RootedTraceableBox;
20use crate::dom::globalscope::GlobalScope;
21use crate::dom::promise::{RootedPromise, TracedPromise};
22use crate::dom::stream::defaultteeunderlyingsource::DefaultTeeUnderlyingSource;
23use crate::dom::stream::readablestream::ReadableStream;
24use crate::realms::enter_auto_realm;
25use crate::runtime::job_queue::MicrotaskRunnable;
26
27#[derive(JSTraceable, MallocSizeOf)]
28#[cfg_attr(crown, expect(crown::unrooted_must_root))]
29pub(crate) struct DefaultTeeReadRequestMicrotask {
30    #[ignore_malloc_size_of = "mozjs"]
31    chunk: Box<Heap<JSVal>>,
32    tee_read_request: Dom<DefaultTeeReadRequest>,
33}
34
35impl MicrotaskRunnable for DefaultTeeReadRequestMicrotask {
36    fn handler(&self, cx: &mut JSContext) {
37        let mut realm = enter_auto_realm(cx, &*self.tee_read_request);
38        self.tee_read_request.chunk_steps(&mut realm, &self.chunk);
39    }
40}
41
42#[dom_struct]
43/// <https://streams.spec.whatwg.org/#ref-for-read-request%E2%91%A2>
44pub(crate) struct DefaultTeeReadRequest {
45    reflector_: Reflector,
46    stream: Dom<ReadableStream>,
47    branch_1: Dom<ReadableStream>,
48    branch_2: Dom<ReadableStream>,
49    #[conditional_malloc_size_of]
50    reading: Rc<Cell<bool>>,
51    #[conditional_malloc_size_of]
52    read_again: Rc<Cell<bool>>,
53    #[conditional_malloc_size_of]
54    canceled_1: Rc<Cell<bool>>,
55    #[conditional_malloc_size_of]
56    canceled_2: Rc<Cell<bool>>,
57    #[conditional_malloc_size_of]
58    clone_for_branch_2: Rc<Cell<bool>>,
59    cancel_promise: TracedPromise,
60    tee_underlying_source: Dom<DefaultTeeUnderlyingSource>,
61}
62impl DefaultTeeReadRequest {
63    #[expect(clippy::too_many_arguments)]
64    pub(crate) fn new(
65        cx: &mut JSContext,
66        stream: &ReadableStream,
67        branch_1: &ReadableStream,
68        branch_2: &ReadableStream,
69        reading: Rc<Cell<bool>>,
70        read_again: Rc<Cell<bool>>,
71        canceled_1: Rc<Cell<bool>>,
72        canceled_2: Rc<Cell<bool>>,
73        clone_for_branch_2: Rc<Cell<bool>>,
74        cancel_promise: &RootedPromise,
75        tee_underlying_source: &DefaultTeeUnderlyingSource,
76    ) -> DomRoot<Self> {
77        reflect_dom_object(
78            cx,
79            Box::new(DefaultTeeReadRequest {
80                reflector_: Reflector::new(),
81                stream: Dom::from_ref(stream),
82                branch_1: Dom::from_ref(branch_1),
83                branch_2: Dom::from_ref(branch_2),
84                reading,
85                read_again,
86                canceled_1,
87                canceled_2,
88                clone_for_branch_2,
89                cancel_promise: cancel_promise.to_traced(),
90                tee_underlying_source: Dom::from_ref(tee_underlying_source),
91            }),
92            &*stream.global(),
93        )
94    }
95    /// Call into cancel of the stream,
96    /// <https://streams.spec.whatwg.org/#readable-stream-cancel>
97    pub(crate) fn stream_cancel(
98        &self,
99        cx: &mut JSContext,
100        global: &GlobalScope,
101        reason: SafeHandleValue,
102    ) {
103        self.stream.cancel(cx, global, reason);
104    }
105    /// Enqueue a microtask to perform the chunk steps
106    /// <https://streams.spec.whatwg.org/#ref-for-read-request-chunk-steps%E2%91%A2>
107    pub(crate) fn enqueue_chunk_steps(
108        &self,
109        cx: &mut JSContext,
110        chunk: RootedTraceableBox<Heap<JSVal>>,
111    ) {
112        // Queue a microtask to perform the following steps:
113        let tee_read_request_chunk = DefaultTeeReadRequestMicrotask {
114            chunk: Heap::boxed(*chunk.handle()),
115            tee_read_request: Dom::from_ref(self),
116        };
117        self.stream
118            .global()
119            .enqueue_microtask(cx, Box::new(tee_read_request_chunk));
120    }
121    /// <https://streams.spec.whatwg.org/#ref-for-read-request-chunk-steps%E2%91%A2>
122    #[expect(clippy::borrowed_box)]
123    pub(crate) fn chunk_steps(&self, cx: &mut JSContext, chunk: &Box<Heap<JSVal>>) {
124        let global = &self.stream.global();
125        // Set readAgain to false.
126        self.read_again.set(false);
127        // Let chunk1 and chunk2 be chunk.
128        rooted!(&in(cx) let chunk1_value = chunk.get());
129        rooted!(&in(cx) let mut chunk2_value = chunk.get());
130        // If canceled_2 is false and cloneForBranch2 is true,
131        if !self.canceled_2.get() && self.clone_for_branch_2.get() {
132            // Let cloneResult be StructuredClone(chunk2).
133            let data = match structuredclone::write(cx, chunk2_value.handle(), None) {
134                Ok(data) => data,
135                Err(error) => {
136                    // If cloneResult is an abrupt completion,
137                    rooted!(&in(cx) let mut error_value = UndefinedValue());
138                    error.to_jsval(cx, global, error_value.handle_mut());
139                    // Perform ! ReadableStreamDefaultControllerError(branch_1.[[controller]], cloneResult.[[Value]]).
140                    self.readable_stream_default_controller_error(
141                        cx,
142                        &self.branch_1,
143                        error_value.handle(),
144                    );
145
146                    // Perform ! ReadableStreamDefaultControllerError(branch_2.[[controller]], cloneResult.[[Value]]).
147                    self.readable_stream_default_controller_error(
148                        cx,
149                        &self.branch_2,
150                        error_value.handle(),
151                    );
152                    // Resolve cancelPromise with ! ReadableStreamCancel(stream, cloneResult.[[Value]]).
153                    self.stream_cancel(cx, global, error_value.handle());
154                    // Return.
155                    return;
156                },
157            };
158            // If cloneResult is an abrupt completion,
159            if let Err(error) = structuredclone::read(cx, global, data, chunk2_value.handle_mut()) {
160                rooted!(&in(cx) let mut error_value = UndefinedValue());
161                error.to_jsval(cx, global, error_value.handle_mut());
162                // Perform ! ReadableStreamDefaultControllerError(branch_1.[[controller]], cloneResult.[[Value]]).
163                self.readable_stream_default_controller_error(
164                    cx,
165                    &self.branch_1,
166                    error_value.handle(),
167                );
168
169                // Perform ! ReadableStreamDefaultControllerError(branch_2.[[controller]], cloneResult.[[Value]]).
170                self.readable_stream_default_controller_error(
171                    cx,
172                    &self.branch_2,
173                    error_value.handle(),
174                );
175                // Resolve cancelPromise with ! ReadableStreamCancel(stream, cloneResult.[[Value]]).
176                self.stream_cancel(cx, global, error_value.handle());
177                // Return.
178                return;
179            }
180        }
181        // If canceled_1 is false, perform ! ReadableStreamDefaultControllerEnqueue(branch_1.[[controller]], chunk1).
182        if !self.canceled_1.get() {
183            self.readable_stream_default_controller_enqueue(
184                cx,
185                &self.branch_1,
186                chunk1_value.handle(),
187            );
188        }
189        // If canceled_2 is false, perform ! ReadableStreamDefaultControllerEnqueue(branch_2.[[controller]], chunk2).
190        if !self.canceled_2.get() {
191            self.readable_stream_default_controller_enqueue(
192                cx,
193                &self.branch_2,
194                chunk2_value.handle(),
195            );
196        }
197        // Set reading to false.
198        self.reading.set(false);
199        // If readAgain is true, perform pullAlgorithm.
200        if self.read_again.get() {
201            self.pull_algorithm(cx);
202        }
203    }
204    /// <https://streams.spec.whatwg.org/#read-request-close-steps>
205    pub(crate) fn close_steps(&self, cx: &mut JSContext) {
206        // Set reading to false.
207        self.reading.set(false);
208        // If canceled_1 is false, perform ! ReadableStreamDefaultControllerClose(branch_1.[[controller]]).
209        if !self.canceled_1.get() {
210            self.readable_stream_default_controller_close(cx, &self.branch_1);
211        }
212        // If canceled_2 is false, perform ! ReadableStreamDefaultControllerClose(branch_2.[[controller]]).
213        if !self.canceled_2.get() {
214            self.readable_stream_default_controller_close(cx, &self.branch_2);
215        }
216        // If canceled_1 is false or canceled_2 is false, resolve cancelPromise with undefined.
217        if !self.canceled_1.get() || !self.canceled_2.get() {
218            self.cancel_promise.resolve_native(cx, &());
219        }
220    }
221    /// <https://streams.spec.whatwg.org/#read-request-error-steps>
222    pub(crate) fn error_steps(&self) {
223        // Set reading to false.
224        self.reading.set(false);
225    }
226    /// Call into enqueue of the default controller of a stream,
227    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-enqueue>
228    fn readable_stream_default_controller_enqueue(
229        &self,
230        cx: &mut JSContext,
231        stream: &ReadableStream,
232        chunk: SafeHandleValue,
233    ) {
234        stream
235            .get_default_controller()
236            .enqueue(cx, chunk)
237            .expect("enqueue failed for stream controller in DefaultTeeReadRequest");
238    }
239
240    /// Call into close of the default controller of a stream,
241    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-close>
242    fn readable_stream_default_controller_close(
243        &self,
244        cx: &mut JSContext,
245        stream: &ReadableStream,
246    ) {
247        stream.get_default_controller().close(cx);
248    }
249
250    /// Call into error of the default controller of stream,
251    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-error>
252    fn readable_stream_default_controller_error(
253        &self,
254        cx: &mut JSContext,
255        stream: &ReadableStream,
256        error: SafeHandleValue,
257    ) {
258        stream.get_default_controller().error(cx, error);
259    }
260
261    pub(crate) fn pull_algorithm(&self, cx: &mut JSContext) {
262        self.tee_underlying_source.pull_algorithm(cx);
263    }
264}
script/dom/stream/defaultteereadrequest.rs

script/dom/stream/
defaultteereadrequest.rs
