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

script/dom/stream/
byteteereadrequest.rs
