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};
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::{RootedPromise, TracedPromise};
23use crate::dom::stream::byteteeunderlyingsource::ByteTeeUnderlyingSource;
24use crate::dom::stream::readablestream::ReadableStream;
25use crate::runtime::job_queue::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    cancel_promise: TracedPromise,
61    tee_underlying_source: Dom<ByteTeeUnderlyingSource>,
62}
63impl ByteTeeReadIntoRequest {
64    #[allow(clippy::too_many_arguments)]
65    pub(crate) fn new(
66        cx: &mut JSContext,
67        for_branch2: bool,
68        byob_branch: &ReadableStream,
69        other_branch: &ReadableStream,
70        stream: &ReadableStream,
71        read_again_for_branch_1: Rc<Cell<bool>>,
72        read_again_for_branch_2: Rc<Cell<bool>>,
73        reading: Rc<Cell<bool>>,
74        canceled_1: Rc<Cell<bool>>,
75        canceled_2: Rc<Cell<bool>>,
76        cancel_promise: &RootedPromise,
77        tee_underlying_source: &ByteTeeUnderlyingSource,
78        global: &GlobalScope,
79    ) -> DomRoot<Self> {
80        reflect_dom_object(
81            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: cancel_promise.to_traced(),
94                tee_underlying_source: Dom::from_ref(tee_underlying_source),
95            }),
96            global,
97        )
98    }
99
100    pub(crate) fn enqueue_chunk_steps(
101        &self,
102        cx: &mut JSContext,
103        chunk: RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>,
104    ) {
105        // Queue a microtask to perform the following steps:
106        let byte_tee_read_request_chunk = ByteTeeReadIntoRequestMicrotask {
107            chunk,
108            tee_read_request: Trusted::new(self),
109        };
110
111        self.global()
112            .enqueue_microtask(cx, Box::new(byte_tee_read_request_chunk));
113    }
114
115    /// <https://streams.spec.whatwg.org/#ref-for-read-into-request-chunk-steps%E2%91%A0>
116    #[allow(clippy::borrowed_box)]
117    pub(crate) fn chunk_steps(
118        &self,
119        chunk: &HeapBufferSource<ArrayBufferViewU8>,
120        cx: &mut JSContext,
121    ) -> Fallible<()> {
122        // Set readAgainForBranch1 to false.
123        self.read_again_for_branch_1.set(false);
124
125        // Set readAgainForBranch2 to false.
126        self.read_again_for_branch_2.set(false);
127
128        // Let byobCanceled be canceled2 if forBranch2 is true, and canceled1 otherwise.
129        let byob_canceled = if self.for_branch2 {
130            self.canceled_2.get()
131        } else {
132            self.canceled_1.get()
133        };
134
135        // Let otherCanceled be canceled2 if forBranch2 is false, and canceled1 otherwise.
136        let other_canceled = if self.for_branch2 {
137            self.canceled_1.get()
138        } else {
139            self.canceled_2.get()
140        };
141
142        // If otherCanceled is false,
143        if !other_canceled {
144            // Let cloneResult be CloneAsUint8Array(chunk).
145            let clone_result = chunk.clone_as_uint8_array(cx);
146
147            // If cloneResult is an abrupt completion,
148            if let Err(error) = clone_result {
149                rooted!(&in(cx) let mut error_value = UndefinedValue());
150                error.to_jsval(cx, &self.global(), error_value.handle_mut());
151
152                // Perform ! ReadableByteStreamControllerError(byobBranch.[[controller]], cloneResult.[[Value]]).
153                let byob_branch_controller = self.byob_branch.get_byte_controller();
154                byob_branch_controller.error(cx, error_value.handle());
155
156                // Perform ! ReadableByteStreamControllerError(otherBranch.[[controller]], cloneResult.[[Value]]).
157                let other_branch_controller = self.other_branch.get_byte_controller();
158                other_branch_controller.error(cx, error_value.handle());
159
160                // Resolve cancelPromise with ! ReadableStreamCancel(stream, cloneResult.[[Value]]).
161                let cancel_result =
162                    self.stream
163                        .cancel(cx, &self.stream.global(), error_value.handle());
164                self.cancel_promise.resolve_native(cx, &cancel_result);
165
166                // Return.
167                return Ok(());
168            } else {
169                // Otherwise, let clonedChunk be cloneResult.[[Value]].
170                let cloned_chunk = clone_result.unwrap();
171
172                // If byobCanceled is false, perform !
173                // ReadableByteStreamControllerRespondWithNewView(byobBranch.[[controller]], chunk).
174                if !byob_canceled {
175                    let byob_branch_controller = self.byob_branch.get_byte_controller();
176                    byob_branch_controller.respond_with_new_view(cx, chunk)?;
177                }
178
179                // Perform ! ReadableByteStreamControllerEnqueue(otherBranch.[[controller]], clonedChunk).
180                let other_branch_controller = self.other_branch.get_byte_controller();
181                other_branch_controller.enqueue(cx, cloned_chunk)?;
182            }
183        } else if !byob_canceled {
184            // Otherwise, if byobCanceled is false, perform
185            // ! ReadableByteStreamControllerRespondWithNewView(byobBranch.[[controller]], chunk).
186
187            let byob_branch_controller = self.byob_branch.get_byte_controller();
188            byob_branch_controller.respond_with_new_view(cx, chunk)?;
189        }
190
191        // Set reading to false.
192        self.reading.set(false);
193
194        // If readAgainForBranch1 is true, perform pull1Algorithm.
195        if self.read_again_for_branch_1.get() {
196            self.pull_algorithm(cx, Some(ByteTeePullAlgorithm::Pull1Algorithm));
197        } else if self.read_again_for_branch_2.get() {
198            // Otherwise, if readAgainForBranch2 is true, perform pull2Algorithm.
199            self.pull_algorithm(cx, Some(ByteTeePullAlgorithm::Pull2Algorithm));
200        }
201
202        Ok(())
203    }
204
205    /// <https://streams.spec.whatwg.org/#ref-for-read-into-request-close-steps%E2%91%A1>
206    pub(crate) fn close_steps(
207        &self,
208        cx: &mut JSContext,
209        chunk: Option<RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>>,
210    ) -> Fallible<()> {
211        // Set reading to false.
212        self.reading.set(false);
213
214        // Let byobCanceled be canceled2 if forBranch2 is true, and canceled1 otherwise.
215        let byob_canceled = if self.for_branch2 {
216            self.canceled_2.get()
217        } else {
218            self.canceled_1.get()
219        };
220
221        // Let otherCanceled be canceled2 if forBranch2 is false, and canceled1 otherwise.
222        let other_canceled = if self.for_branch2 {
223            self.canceled_1.get()
224        } else {
225            self.canceled_2.get()
226        };
227
228        // If byobCanceled is false, perform ! ReadableByteStreamControllerClose(byobBranch.[[controller]]).
229        if !byob_canceled {
230            let byob_branch_controller = self.byob_branch.get_byte_controller();
231            byob_branch_controller.close(cx)?;
232        }
233
234        // If otherCanceled is false, perform ! ReadableByteStreamControllerClose(otherBranch.[[controller]]).
235        if !other_canceled {
236            let other_branch_controller = self.other_branch.get_byte_controller();
237            other_branch_controller.close(cx)?;
238        }
239
240        // If chunk is not undefined,
241        if let Some(chunk_value) = chunk {
242            if chunk_value.is_undefined() {
243                // Nothing to respond with if the provided chunk is undefined.
244                // Continue with the remaining close steps.
245            } else {
246                let chunk = chunk_value;
247                // Assert: chunk.[[ByteLength]] is 0.
248                assert_eq!(chunk.byte_length(), 0);
249
250                // If byobCanceled is false, perform !
251                // ReadableByteStreamControllerRespondWithNewView(byobBranch.[[controller]], chunk).
252                if !byob_canceled {
253                    let byob_branch_controller = self.byob_branch.get_byte_controller();
254                    byob_branch_controller.respond_with_new_view(cx, &chunk)?;
255                }
256
257                // If otherCanceled is false and otherBranch.[[controller]].[[pendingPullIntos]] is not empty,
258                // perform ! ReadableByteStreamControllerRespond(otherBranch.[[controller]], 0).
259                if !other_canceled {
260                    let other_branch_controller = self.other_branch.get_byte_controller();
261                    if other_branch_controller.get_pending_pull_intos_size() > 0 {
262                        other_branch_controller.respond(cx, 0)?;
263                    }
264                }
265            }
266        }
267
268        // If byobCanceled is false or otherCanceled is false, resolve cancelPromise with undefined.
269        if !byob_canceled || !other_canceled {
270            self.cancel_promise.resolve_native(cx, &());
271        }
272
273        Ok(())
274    }
275    /// <https://streams.spec.whatwg.org/#ref-for-read-into-request-error-steps%E2%91%A0>
276    pub(crate) fn error_steps(&self) {
277        // Set reading to false.
278        self.reading.set(false);
279    }
280
281    pub(crate) fn pull_algorithm(
282        &self,
283        cx: &mut JSContext,
284        byte_tee_pull_algorithm: Option<ByteTeePullAlgorithm>,
285    ) {
286        self.tee_underlying_source
287            .pull_algorithm(cx, byte_tee_pull_algorithm);
288    }
289}
script/dom/stream/byteteereadintorequest.rs

script/dom/stream/
byteteereadintorequest.rs
