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

script/dom/stream/
byteteereadrequest.rs
