script/dom/stream/
byteteereadrequest.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
5#![cfg_attr(crown, allow(crown::jscontext_first_arg))]
6
7use std::cell::Cell;
8use std::rc::Rc;
9
10use dom_struct::dom_struct;
11use js::context::JSContext;
12use js::jsapi::Heap;
13use js::jsval::{JSVal, UndefinedValue};
14use js::typedarray::ArrayBufferViewU8;
15use script_bindings::error::Fallible;
16use script_bindings::reflector::{Reflector, reflect_dom_object};
17
18use super::byteteeunderlyingsource::ByteTeePullAlgorithm;
19use crate::dom::bindings::buffer_source::HeapBufferSource;
20use crate::dom::bindings::error::{Error, ErrorToJsval};
21use crate::dom::bindings::reflector::DomGlobal;
22use crate::dom::bindings::root::{Dom, DomRoot};
23use crate::dom::bindings::trace::RootedTraceableBox;
24use crate::dom::globalscope::GlobalScope;
25use crate::dom::promise::{RootedPromise, TracedPromise};
26use crate::dom::stream::byteteeunderlyingsource::ByteTeeUnderlyingSource;
27use crate::dom::stream::readablestream::ReadableStream;
28use crate::realms::enter_auto_realm;
29use crate::runtime::job_queue::MicrotaskRunnable;
30
31#[derive(JSTraceable, MallocSizeOf)]
32#[cfg_attr(crown, expect(crown::unrooted_must_root))]
33pub(crate) struct ByteTeeReadRequestMicrotask {
34    #[ignore_malloc_size_of = "mozjs"]
35    chunk: Box<Heap<JSVal>>,
36    tee_read_request: Dom<ByteTeeReadRequest>,
37}
38
39impl MicrotaskRunnable for ByteTeeReadRequestMicrotask {
40    fn handler(&self, cx: &mut JSContext) {
41        let mut realm = enter_auto_realm(cx, &*self.tee_read_request);
42        let cx = &mut realm;
43        self.tee_read_request
44            .chunk_steps(&self.chunk, cx)
45            .expect("ByteTeeReadRequestMicrotask::microtask_chunk_steps failed");
46    }
47}
48
49#[dom_struct]
50/// <https://streams.spec.whatwg.org/#ref-for-read-request%E2%91%A2>
51pub(crate) struct ByteTeeReadRequest {
52    reflector_: Reflector,
53    branch_1: Dom<ReadableStream>,
54    branch_2: Dom<ReadableStream>,
55    stream: Dom<ReadableStream>,
56    #[conditional_malloc_size_of]
57    read_again_for_branch_1: Rc<Cell<bool>>,
58    #[conditional_malloc_size_of]
59    read_again_for_branch_2: Rc<Cell<bool>>,
60    #[conditional_malloc_size_of]
61    reading: Rc<Cell<bool>>,
62    #[conditional_malloc_size_of]
63    canceled_1: Rc<Cell<bool>>,
64    #[conditional_malloc_size_of]
65    canceled_2: Rc<Cell<bool>>,
66    cancel_promise: TracedPromise,
67    tee_underlying_source: Dom<ByteTeeUnderlyingSource>,
68}
69impl ByteTeeReadRequest {
70    #[allow(clippy::too_many_arguments)]
71    pub(crate) fn new(
72        cx: &mut JSContext,
73        branch_1: &ReadableStream,
74        branch_2: &ReadableStream,
75        stream: &ReadableStream,
76        read_again_for_branch_1: Rc<Cell<bool>>,
77        read_again_for_branch_2: Rc<Cell<bool>>,
78        reading: Rc<Cell<bool>>,
79        canceled_1: Rc<Cell<bool>>,
80        canceled_2: Rc<Cell<bool>>,
81        cancel_promise: &RootedPromise,
82        tee_underlying_source: &ByteTeeUnderlyingSource,
83        global: &GlobalScope,
84    ) -> DomRoot<Self> {
85        reflect_dom_object(
86            cx,
87            Box::new(ByteTeeReadRequest {
88                reflector_: Reflector::new(),
89                branch_1: Dom::from_ref(branch_1),
90                branch_2: Dom::from_ref(branch_2),
91                stream: Dom::from_ref(stream),
92                read_again_for_branch_1,
93                read_again_for_branch_2,
94                reading,
95                canceled_1,
96                canceled_2,
97                cancel_promise: cancel_promise.to_traced(),
98                tee_underlying_source: Dom::from_ref(tee_underlying_source),
99            }),
100            global,
101        )
102    }
103
104    /// Enqueue a microtask to perform the chunk steps
105    /// <https://streams.spec.whatwg.org/#ref-for-read-request-chunk-steps%E2%91%A2>
106    pub(crate) fn enqueue_chunk_steps(
107        &self,
108        cx: &mut JSContext,
109        global: &GlobalScope,
110        chunk: RootedTraceableBox<Heap<JSVal>>,
111    ) {
112        // Queue a microtask to perform the following steps:
113        let byte_tee_read_request_chunk = ByteTeeReadRequestMicrotask {
114            chunk: Heap::boxed(*chunk.handle()),
115            tee_read_request: Dom::from_ref(self),
116        };
117        global.enqueue_microtask(cx, Box::new(byte_tee_read_request_chunk));
118    }
119
120    /// <https://streams.spec.whatwg.org/#ref-for-read-request-chunk-steps%E2%91%A3>
121    #[allow(clippy::borrowed_box)]
122    pub(crate) fn chunk_steps(&self, chunk: &Box<Heap<JSVal>>, cx: &mut JSContext) -> Fallible<()> {
123        // Set readAgainForBranch1 to false.
124        self.read_again_for_branch_1.set(false);
125
126        // Set readAgainForBranch2 to false.
127        self.read_again_for_branch_2.set(false);
128
129        // Let chunk1 and chunk2 be chunk.
130        rooted!(&in(cx) let chunk_object = chunk.get().to_object());
131
132        // Helper to surface clone failures exactly once
133        let handle_clone_error = |cx: &mut JSContext, error: Error| {
134            rooted!(&in(cx) let mut error_value = UndefinedValue());
135            error.to_jsval(cx, &self.global(), error_value.handle_mut());
136
137            let branch_1_controller = self.branch_1.get_byte_controller();
138            let branch_2_controller = self.branch_2.get_byte_controller();
139
140            branch_1_controller.error(cx, error_value.handle());
141            branch_2_controller.error(cx, error_value.handle());
142
143            let cancel_result = self
144                .stream
145                .cancel(cx, &self.stream.global(), error_value.handle());
146            self.cancel_promise.resolve_native(cx, &cancel_result);
147        };
148
149        // Prepare per branch chunks ahead of the spec enqueue steps.
150        let chunk1_view = if !self.canceled_1.get() {
151            Some(RootedTraceableBox::new(
152                HeapBufferSource::<ArrayBufferViewU8>::new(chunk_object.handle()),
153            ))
154        } else {
155            None
156        };
157
158        let mut chunk2_view = None;
159
160        // If canceled1 is false and canceled2 is false,
161        if !self.canceled_1.get() && !self.canceled_2.get() {
162            // Let cloneResult be CloneAsUint8Array(chunk).
163            let chunk2_source = RootedTraceableBox::new(
164                HeapBufferSource::<ArrayBufferViewU8>::new(chunk_object.handle()),
165            );
166            let clone_result = chunk2_source.clone_as_uint8_array(cx);
167
168            // If cloneResult is an abrupt completion,
169            if let Err(error) = clone_result {
170                handle_clone_error(cx, error);
171                return Ok(());
172            } else {
173                // Otherwise, set chunk2 to cloneResult.[[Value]].
174                chunk2_view = clone_result.ok();
175            }
176        } else if !self.canceled_2.get() {
177            // Only branch2 needs data; clone once for it.
178            let chunk2_source = RootedTraceableBox::new(
179                HeapBufferSource::<ArrayBufferViewU8>::new(chunk_object.handle()),
180            );
181            match chunk2_source.clone_as_uint8_array(cx) {
182                Ok(clone) => chunk2_view = Some(clone),
183                Err(error) => {
184                    handle_clone_error(cx, error);
185                    return Ok(());
186                },
187            }
188        }
189
190        // If canceled1 is false, perform ! ReadableByteStreamControllerEnqueue(branch1.[[controller]], chunk1).
191        if let Some(chunk1_view) = chunk1_view {
192            let branch_1_controller = self.branch_1.get_byte_controller();
193            branch_1_controller.enqueue(cx, chunk1_view)?;
194        }
195
196        // If canceled2 is false, perform ! ReadableByteStreamControllerEnqueue(branch2.[[controller]], chunk2).
197        if let Some(chunk2_view) = chunk2_view {
198            let branch_2_controller = self.branch_2.get_byte_controller();
199            branch_2_controller.enqueue(cx, chunk2_view)?;
200        }
201
202        // Set reading to false.
203        self.reading.set(false);
204
205        // If readAgainForBranch1 is true, perform pull1Algorithm.
206        if self.read_again_for_branch_1.get() {
207            self.pull_algorithm(cx, Some(ByteTeePullAlgorithm::Pull1Algorithm));
208        } else if self.read_again_for_branch_2.get() {
209            // Otherwise, if readAgainForBranch2 is true, perform pull2Algorithm.
210            self.pull_algorithm(cx, Some(ByteTeePullAlgorithm::Pull2Algorithm));
211        }
212
213        Ok(())
214    }
215
216    /// <https://streams.spec.whatwg.org/#ref-for-read-request-close-steps%E2%91%A2>
217    pub(crate) fn close_steps(&self, cx: &mut JSContext) -> Fallible<()> {
218        let branch_1_controller = self.branch_1.get_byte_controller();
219        let branch_2_controller = self.branch_2.get_byte_controller();
220
221        // Set reading to false.
222        self.reading.set(false);
223
224        // If canceled1 is false, perform ! ReadableByteStreamControllerClose(branch1.[[controller]]).
225        if !self.canceled_1.get() {
226            branch_1_controller.close(cx)?;
227        }
228
229        // If canceled2 is false, perform ! ReadableByteStreamControllerClose(branch2.[[controller]]).
230        if !self.canceled_2.get() {
231            branch_2_controller.close(cx)?;
232        }
233
234        // If branch1.[[controller]].[[pendingPullIntos]] is not empty,
235        // perform ! ReadableByteStreamControllerRespond(branch1.[[controller]], 0).
236        if branch_1_controller.get_pending_pull_intos_size() > 0 {
237            branch_1_controller.respond(cx, 0)?;
238        }
239
240        // If branch2.[[controller]].[[pendingPullIntos]] is not empty,
241        // perform ! ReadableByteStreamControllerRespond(branch2.[[controller]], 0).
242        if branch_2_controller.get_pending_pull_intos_size() > 0 {
243            branch_2_controller.respond(cx, 0)?;
244        }
245
246        // If canceled1 is false or canceled2 is false, resolve cancelPromise with undefined.
247        if !self.canceled_1.get() || !self.canceled_2.get() {
248            self.cancel_promise.resolve_native(cx, &());
249        }
250
251        Ok(())
252    }
253
254    /// <https://streams.spec.whatwg.org/#ref-for-read-request-error-steps%E2%91%A3>
255    pub(crate) fn error_steps(&self) {
256        // Set reading to false.
257        self.reading.set(false);
258    }
259
260    pub(crate) fn pull_algorithm(
261        &self,
262        cx: &mut JSContext,
263        byte_tee_pull_algorithm: Option<ByteTeePullAlgorithm>,
264    ) {
265        self.tee_underlying_source
266            .pull_algorithm(cx, byte_tee_pull_algorithm);
267    }
268}
script/dom/stream/byteteereadrequest.rs

script/dom/stream/
byteteereadrequest.rs
