script/dom/stream/
byteteereadintorequest.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::jsval::UndefinedValue;
11use js::typedarray::ArrayBufferViewU8;
12use script_bindings::reflector::{Reflector, reflect_dom_object_with_cx};
13use script_bindings::trace::RootedTraceableBox;
14
15use super::byteteeunderlyingsource::ByteTeePullAlgorithm;
16use crate::dom::bindings::buffer_source::HeapBufferSource;
17use crate::dom::bindings::error::{ErrorToJsval, Fallible};
18use crate::dom::bindings::refcounted::Trusted;
19use crate::dom::bindings::reflector::DomGlobal;
20use crate::dom::bindings::root::{Dom, DomRoot};
21use crate::dom::globalscope::GlobalScope;
22use crate::dom::promise::Promise;
23use crate::dom::stream::byteteeunderlyingsource::ByteTeeUnderlyingSource;
24use crate::dom::stream::readablestream::ReadableStream;
25use crate::microtask::MicrotaskRunnable;
26
27#[derive(JSTraceable, MallocSizeOf)]
28pub(crate) struct ByteTeeReadIntoRequestMicrotask {
29    #[ignore_malloc_size_of = "mozjs"]
30    chunk: RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>,
31    tee_read_request: Trusted<ByteTeeReadIntoRequest>,
32}
33
34impl MicrotaskRunnable for ByteTeeReadIntoRequestMicrotask {
35    fn handler(&self, cx: &mut JSContext) {
36        self.tee_read_request
37            .root()
38            .chunk_steps(&self.chunk, cx)
39            .expect("Failed to enqueue chunk");
40    }
41}
42
43#[dom_struct]
44pub(crate) struct ByteTeeReadIntoRequest {
45    reflector_: Reflector,
46    for_branch2: bool,
47    byob_branch: Dom<ReadableStream>,
48    other_branch: Dom<ReadableStream>,
49    stream: Dom<ReadableStream>,
50    #[conditional_malloc_size_of]
51    read_again_for_branch_1: Rc<Cell<bool>>,
52    #[conditional_malloc_size_of]
53    read_again_for_branch_2: Rc<Cell<bool>>,
54    #[conditional_malloc_size_of]
55    reading: Rc<Cell<bool>>,
56    #[conditional_malloc_size_of]
57    canceled_1: Rc<Cell<bool>>,
58    #[conditional_malloc_size_of]
59    canceled_2: Rc<Cell<bool>>,
60    #[conditional_malloc_size_of]
61    cancel_promise: Rc<Promise>,
62    tee_underlying_source: Dom<ByteTeeUnderlyingSource>,
63}
64impl ByteTeeReadIntoRequest {
65    #[allow(clippy::too_many_arguments)]
66    pub(crate) fn new(
67        cx: &mut JSContext,
68        for_branch2: bool,
69        byob_branch: &ReadableStream,
70        other_branch: &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(ByteTeeReadIntoRequest {
83                reflector_: Reflector::new(),
84                for_branch2,
85                byob_branch: Dom::from_ref(byob_branch),
86                other_branch: Dom::from_ref(other_branch),
87                stream: Dom::from_ref(stream),
88                read_again_for_branch_1,
89                read_again_for_branch_2,
90                reading,
91                canceled_1,
92                canceled_2,
93                cancel_promise,
94                tee_underlying_source: Dom::from_ref(tee_underlying_source),
95            }),
96            global,
97            cx,
98        )
99    }
100
101    pub(crate) fn enqueue_chunk_steps(
102        &self,
103        cx: &mut JSContext,
104        chunk: RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>,
105    ) {
106        // Queue a microtask to perform the following steps:
107        let byte_tee_read_request_chunk = ByteTeeReadIntoRequestMicrotask {
108            chunk,
109            tee_read_request: Trusted::new(self),
110        };
111
112        self.global()
113            .enqueue_microtask(cx, Box::new(byte_tee_read_request_chunk));
114    }
115
116    /// <https://streams.spec.whatwg.org/#ref-for-read-into-request-chunk-steps%E2%91%A0>
117    #[allow(clippy::borrowed_box)]
118    pub(crate) fn chunk_steps(
119        &self,
120        chunk: &HeapBufferSource<ArrayBufferViewU8>,
121        cx: &mut JSContext,
122    ) -> 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 byobCanceled be canceled2 if forBranch2 is true, and canceled1 otherwise.
130        let byob_canceled = if self.for_branch2 {
131            self.canceled_2.get()
132        } else {
133            self.canceled_1.get()
134        };
135
136        // Let otherCanceled be canceled2 if forBranch2 is false, and canceled1 otherwise.
137        let other_canceled = if self.for_branch2 {
138            self.canceled_1.get()
139        } else {
140            self.canceled_2.get()
141        };
142
143        // If otherCanceled is false,
144        if !other_canceled {
145            // Let cloneResult be CloneAsUint8Array(chunk).
146            let clone_result = chunk.clone_as_uint8_array(cx);
147
148            // If cloneResult is an abrupt completion,
149            if let Err(error) = clone_result {
150                rooted!(&in(cx) let mut error_value = UndefinedValue());
151                error.to_jsval(cx, &self.global(), error_value.handle_mut());
152
153                // Perform ! ReadableByteStreamControllerError(byobBranch.[[controller]], cloneResult.[[Value]]).
154                let byob_branch_controller = self.byob_branch.get_byte_controller();
155                byob_branch_controller.error(cx, error_value.handle());
156
157                // Perform ! ReadableByteStreamControllerError(otherBranch.[[controller]], cloneResult.[[Value]]).
158                let other_branch_controller = self.other_branch.get_byte_controller();
159                other_branch_controller.error(cx, error_value.handle());
160
161                // Resolve cancelPromise with ! ReadableStreamCancel(stream, cloneResult.[[Value]]).
162                let cancel_result =
163                    self.stream
164                        .cancel(cx, &self.stream.global(), error_value.handle());
165                self.cancel_promise.resolve_native(cx, &cancel_result);
166
167                // Return.
168                return Ok(());
169            } else {
170                // Otherwise, let clonedChunk be cloneResult.[[Value]].
171                let cloned_chunk = clone_result.unwrap();
172
173                // If byobCanceled is false, perform !
174                // ReadableByteStreamControllerRespondWithNewView(byobBranch.[[controller]], chunk).
175                if !byob_canceled {
176                    let byob_branch_controller = self.byob_branch.get_byte_controller();
177                    byob_branch_controller.respond_with_new_view(cx, chunk)?;
178                }
179
180                // Perform ! ReadableByteStreamControllerEnqueue(otherBranch.[[controller]], clonedChunk).
181                let other_branch_controller = self.other_branch.get_byte_controller();
182                other_branch_controller.enqueue(cx, cloned_chunk)?;
183            }
184        } else if !byob_canceled {
185            // Otherwise, if byobCanceled is false, perform
186            // ! ReadableByteStreamControllerRespondWithNewView(byobBranch.[[controller]], chunk).
187
188            let byob_branch_controller = self.byob_branch.get_byte_controller();
189            byob_branch_controller.respond_with_new_view(cx, chunk)?;
190        }
191
192        // Set reading to false.
193        self.reading.set(false);
194
195        // If readAgainForBranch1 is true, perform pull1Algorithm.
196        if self.read_again_for_branch_1.get() {
197            self.pull_algorithm(cx, Some(ByteTeePullAlgorithm::Pull1Algorithm));
198        } else if self.read_again_for_branch_2.get() {
199            // Otherwise, if readAgainForBranch2 is true, perform pull2Algorithm.
200            self.pull_algorithm(cx, Some(ByteTeePullAlgorithm::Pull2Algorithm));
201        }
202
203        Ok(())
204    }
205
206    /// <https://streams.spec.whatwg.org/#ref-for-read-into-request-close-steps%E2%91%A1>
207    pub(crate) fn close_steps(
208        &self,
209        cx: &mut JSContext,
210        chunk: Option<RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>>,
211    ) -> Fallible<()> {
212        // Set reading to false.
213        self.reading.set(false);
214
215        // Let byobCanceled be canceled2 if forBranch2 is true, and canceled1 otherwise.
216        let byob_canceled = if self.for_branch2 {
217            self.canceled_2.get()
218        } else {
219            self.canceled_1.get()
220        };
221
222        // Let otherCanceled be canceled2 if forBranch2 is false, and canceled1 otherwise.
223        let other_canceled = if self.for_branch2 {
224            self.canceled_1.get()
225        } else {
226            self.canceled_2.get()
227        };
228
229        // If byobCanceled is false, perform ! ReadableByteStreamControllerClose(byobBranch.[[controller]]).
230        if !byob_canceled {
231            let byob_branch_controller = self.byob_branch.get_byte_controller();
232            byob_branch_controller.close(cx)?;
233        }
234
235        // If otherCanceled is false, perform ! ReadableByteStreamControllerClose(otherBranch.[[controller]]).
236        if !other_canceled {
237            let other_branch_controller = self.other_branch.get_byte_controller();
238            other_branch_controller.close(cx)?;
239        }
240
241        // If chunk is not undefined,
242        if let Some(chunk_value) = chunk {
243            if chunk_value.is_undefined() {
244                // Nothing to respond with if the provided chunk is undefined.
245                // Continue with the remaining close steps.
246            } else {
247                let chunk = chunk_value;
248                // Assert: chunk.[[ByteLength]] is 0.
249                assert_eq!(chunk.byte_length(), 0);
250
251                // If byobCanceled is false, perform !
252                // ReadableByteStreamControllerRespondWithNewView(byobBranch.[[controller]], chunk).
253                if !byob_canceled {
254                    let byob_branch_controller = self.byob_branch.get_byte_controller();
255                    byob_branch_controller.respond_with_new_view(cx, &chunk)?;
256                }
257
258                // If otherCanceled is false and otherBranch.[[controller]].[[pendingPullIntos]] is not empty,
259                // perform ! ReadableByteStreamControllerRespond(otherBranch.[[controller]], 0).
260                if !other_canceled {
261                    let other_branch_controller = self.other_branch.get_byte_controller();
262                    if other_branch_controller.get_pending_pull_intos_size() > 0 {
263                        other_branch_controller.respond(cx, 0)?;
264                    }
265                }
266            }
267        }
268
269        // If byobCanceled is false or otherCanceled is false, resolve cancelPromise with undefined.
270        if !byob_canceled || !other_canceled {
271            self.cancel_promise.resolve_native(cx, &());
272        }
273
274        Ok(())
275    }
276    /// <https://streams.spec.whatwg.org/#ref-for-read-into-request-error-steps%E2%91%A0>
277    pub(crate) fn error_steps(&self) {
278        // Set reading to false.
279        self.reading.set(false);
280    }
281
282    pub(crate) fn pull_algorithm(
283        &self,
284        cx: &mut JSContext,
285        byte_tee_pull_algorithm: Option<ByteTeePullAlgorithm>,
286    ) {
287        self.tee_underlying_source
288            .pull_algorithm(cx, byte_tee_pull_algorithm);
289    }
290}
script/dom/stream/byteteereadintorequest.rs

script/dom/stream/
byteteereadintorequest.rs
