script/dom/stream/
byteteeunderlyingsource.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, RefCell};
6use std::rc::Rc;
7
8use dom_struct::dom_struct;
9use js::context::JSContext;
10use js::jsapi::{HandleValueArray, Heap, NewArrayObject, Value};
11use js::jsval::ObjectValue;
12use js::rust::HandleValue as SafeHandleValue;
13use js::typedarray::ArrayBufferViewU8;
14use script_bindings::reflector::{Reflector, reflect_dom_object_with_cx};
15
16use super::byteteereadintorequest::ByteTeeReadIntoRequest;
17use super::readablestream::ReaderType;
18use super::readablestreambyobreader::ReadIntoRequest;
19use crate::dom::bindings::buffer_source::HeapBufferSource;
20use crate::dom::bindings::error::{Error, Fallible};
21use crate::dom::bindings::reflector::DomGlobal;
22use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
23use crate::dom::globalscope::GlobalScope;
24use crate::dom::promise::Promise;
25use crate::dom::stream::byteteereadrequest::ByteTeeReadRequest;
26use crate::dom::stream::readablestreamdefaultreader::ReadRequest;
27use crate::dom::types::ReadableStream;
28
29#[derive(JSTraceable, MallocSizeOf)]
30pub(crate) enum ByteTeeCancelAlgorithm {
31    Cancel1Algorithm,
32    Cancel2Algorithm,
33}
34
35#[derive(Clone, JSTraceable, MallocSizeOf)]
36pub(crate) enum ByteTeePullAlgorithm {
37    Pull1Algorithm,
38    Pull2Algorithm,
39}
40
41#[dom_struct]
42/// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaulttee>
43pub(crate) struct ByteTeeUnderlyingSource {
44    reflector_: Reflector,
45    #[conditional_malloc_size_of]
46    reader: Rc<RefCell<ReaderType>>,
47    stream: Dom<ReadableStream>,
48    branch_1: MutNullableDom<ReadableStream>,
49    branch_2: MutNullableDom<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    #[ignore_malloc_size_of = "Mozjs"]
61    reason_1: Rc<Heap<Value>>,
62    #[ignore_malloc_size_of = "Mozjs"]
63    reason_2: Rc<Heap<Value>>,
64    #[conditional_malloc_size_of]
65    cancel_promise: Rc<Promise>,
66    #[conditional_malloc_size_of]
67    reader_version: Rc<Cell<u64>>,
68    tee_cancel_algorithm: ByteTeeCancelAlgorithm,
69    byte_tee_pull_algorithm: ByteTeePullAlgorithm,
70}
71
72impl ByteTeeUnderlyingSource {
73    #[allow(clippy::too_many_arguments)]
74    #[cfg_attr(crown, expect(crown::unrooted_must_root))]
75    pub(crate) fn new(
76        cx: &mut JSContext,
77        reader: Rc<RefCell<ReaderType>>,
78        stream: &ReadableStream,
79        reading: Rc<Cell<bool>>,
80        read_again_for_branch_1: Rc<Cell<bool>>,
81        read_again_for_branch_2: Rc<Cell<bool>>,
82        canceled_1: Rc<Cell<bool>>,
83        canceled_2: Rc<Cell<bool>>,
84        reason_1: Rc<Heap<Value>>,
85        reason_2: Rc<Heap<Value>>,
86        cancel_promise: Rc<Promise>,
87        reader_version: Rc<Cell<u64>>,
88        tee_cancel_algorithm: ByteTeeCancelAlgorithm,
89        byte_tee_pull_algorithm: ByteTeePullAlgorithm,
90    ) -> DomRoot<ByteTeeUnderlyingSource> {
91        reflect_dom_object_with_cx(
92            Box::new(ByteTeeUnderlyingSource {
93                reflector_: Reflector::new(),
94                reader,
95                stream: Dom::from_ref(stream),
96                branch_1: MutNullableDom::new(None),
97                branch_2: MutNullableDom::new(None),
98                read_again_for_branch_1,
99                read_again_for_branch_2,
100                reading,
101                canceled_1,
102                canceled_2,
103                reason_1,
104                reason_2,
105                cancel_promise,
106                reader_version,
107                tee_cancel_algorithm,
108                byte_tee_pull_algorithm,
109            }),
110            &*stream.global(),
111            cx,
112        )
113    }
114
115    pub(crate) fn set_branch_1(&self, stream: &ReadableStream) {
116        self.branch_1.set(Some(stream));
117    }
118
119    pub(crate) fn set_branch_2(&self, stream: &ReadableStream) {
120        self.branch_2.set(Some(stream));
121    }
122
123    #[cfg_attr(crown, expect(crown::unrooted_must_root))]
124    pub(crate) fn forward_reader_error(
125        &self,
126        cx: &mut JSContext,
127        this_reader: Rc<RefCell<ReaderType>>,
128    ) {
129        let this_reader = this_reader.borrow();
130        match &*this_reader {
131            ReaderType::Default(reader) => {
132                let expected_version = self.reader_version.get();
133                // Upon rejection of thisReader.[[closedPromise]] with reason r,
134                reader
135                    .get()
136                    .expect("Reader should be set.")
137                    .byte_tee_append_native_handler_to_closed_promise(
138                        cx,
139                        &self.branch_1.get().expect("Branch 1 should be set."),
140                        &self.branch_2.get().expect("Branch 2 should be set."),
141                        self.canceled_1.clone(),
142                        self.canceled_2.clone(),
143                        self.cancel_promise.clone(),
144                        self.reader_version.clone(),
145                        expected_version,
146                    );
147            },
148            ReaderType::BYOB(reader) => {
149                let expected_version = self.reader_version.get();
150                // Upon rejection of thisReader.[[closedPromise]] with reason r,
151                reader
152                    .get()
153                    .expect("Reader should be set.")
154                    .byte_tee_append_native_handler_to_closed_promise(
155                        cx,
156                        &self.branch_1.get().expect("Branch 1 should be set."),
157                        &self.branch_2.get().expect("Branch 2 should be set."),
158                        self.canceled_1.clone(),
159                        self.canceled_2.clone(),
160                        self.cancel_promise.clone(),
161                        self.reader_version.clone(),
162                        expected_version,
163                    );
164            },
165        }
166    }
167
168    fn pull_with_default_reader(&self, cx: &mut JSContext, global: &GlobalScope) -> Fallible<()> {
169        rooted!(&in(cx) let mut reader_to_set = None);
170        {
171            let reader = self.reader.borrow();
172            match &*reader {
173                ReaderType::BYOB(byte_reader) => {
174                    // Assert: readIntoRequests is empty.
175                    assert!(
176                        byte_reader
177                            .get()
178                            .expect("Reader should be set.")
179                            .get_num_read_into_requests() ==
180                            0
181                    );
182
183                    // Release BYOB reader.
184                    byte_reader
185                        .get()
186                        .expect("Reader should be set.")
187                        .release(cx)?;
188
189                    // Acquire default reader.
190                    let default_reader = self
191                        .stream
192                        .acquire_default_reader(cx)
193                        .expect("AcquireReadableStreamDefaultReader should not fail");
194
195                    reader_to_set.set(Some(ReaderType::Default(MutNullableDom::new(Some(
196                        &default_reader,
197                    )))));
198                    // We continue after the reader is set, after this match.
199                },
200                ReaderType::Default(reader) => {
201                    let byte_tee_read_request = ByteTeeReadRequest::new(
202                        cx,
203                        &self.branch_1.get().expect("Branch 1 should be set."),
204                        &self.branch_2.get().expect("Branch 2 should be set."),
205                        &self.stream,
206                        self.read_again_for_branch_1.clone(),
207                        self.read_again_for_branch_2.clone(),
208                        self.reading.clone(),
209                        self.canceled_1.clone(),
210                        self.canceled_2.clone(),
211                        self.cancel_promise.clone(),
212                        self,
213                        global,
214                    );
215
216                    let read_request = ReadRequest::ByteTee {
217                        byte_tee_read_request: Dom::from_ref(&byte_tee_read_request),
218                    };
219
220                    reader
221                        .get()
222                        .expect("Reader should be set.")
223                        .read(cx, &read_request);
224                },
225            }
226        }
227
228        if reader_to_set.is_some() {
229            *self.reader.borrow_mut() = reader_to_set.take().unwrap();
230            self.reader_version
231                .set(self.reader_version.get().wrapping_add(1));
232
233            // Attach error forwarding for the new reader.
234            self.forward_reader_error(cx, self.reader.clone());
235
236            // IMPORTANT: now actually perform the pull we were asked to do.
237            return self.pull_with_default_reader(cx, global);
238        }
239
240        Ok(())
241    }
242
243    fn pull_with_byob_reader(
244        &self,
245        cx: &mut JSContext,
246        view: &HeapBufferSource<ArrayBufferViewU8>,
247        for_branch2: bool,
248        global: &GlobalScope,
249    ) {
250        rooted!(&in(cx) let mut reader_to_set = None);
251        {
252            let reader = self.reader.borrow();
253            match &*reader {
254                ReaderType::BYOB(reader) => {
255                    // Let byobBranch be branch2 if forBranch2 is true, and branch1 otherwise.
256                    let byob_branch = if for_branch2 {
257                        self.branch_2.get().expect("Branch 2 should be set.")
258                    } else {
259                        self.branch_1.get().expect("Branch 1 should be set.")
260                    };
261
262                    // let otherBranch be branch2 if forBranch2 is false, and branch1 otherwise.
263                    let other_branch = if for_branch2 {
264                        self.branch_1.get().expect("Branch 1 should be set.")
265                    } else {
266                        self.branch_2.get().expect("Branch 2 should be set.")
267                    };
268
269                    // Let readIntoRequest be a read-into request with the following items:
270                    let byte_tee_read_into_request = ByteTeeReadIntoRequest::new(
271                        cx,
272                        for_branch2,
273                        &byob_branch,
274                        &other_branch,
275                        &self.stream,
276                        self.read_again_for_branch_1.clone(),
277                        self.read_again_for_branch_2.clone(),
278                        self.reading.clone(),
279                        self.canceled_1.clone(),
280                        self.canceled_2.clone(),
281                        self.cancel_promise.clone(),
282                        self,
283                        global,
284                    );
285
286                    let read_into_request = ReadIntoRequest::ByteTee {
287                        byte_tee_read_into_request: Dom::from_ref(&byte_tee_read_into_request),
288                    };
289
290                    // Perform ! ReadableStreamBYOBReaderRead(reader, view, 1, readIntoRequest).
291                    reader.get().expect("Reader should be set.").read(
292                        cx,
293                        view,
294                        1,
295                        &read_into_request,
296                    );
297                },
298                ReaderType::Default(default_reader) => {
299                    // If reader implements ReadableStreamDefaultReader,
300                    // Assert: reader.[[readRequests]] is empty.
301                    assert!(
302                        default_reader
303                            .get()
304                            .expect("Reader should be set.")
305                            .get_num_read_requests() ==
306                            0
307                    );
308
309                    // Perform ! ReadableStreamDefaultReaderRelease(reader).
310                    default_reader
311                        .get()
312                        .expect("Reader should be set.")
313                        .release(cx)
314                        .expect("Release should be successful.");
315
316                    // Set reader to ! AcquireReadableStreamBYOBReader(stream).
317                    let byob_reader = self
318                        .stream
319                        .acquire_byob_reader(cx)
320                        .expect("Reader should be set.");
321
322                    reader_to_set.set(Some(ReaderType::BYOB(MutNullableDom::new(Some(
323                        &byob_reader,
324                    )))));
325                    // This execution path continues after we set the reader.
326                },
327            }
328        }
329
330        if reader_to_set.is_some() {
331            *self.reader.borrow_mut() = reader_to_set.take().unwrap();
332            self.reader_version
333                .set(self.reader_version.get().wrapping_add(1));
334
335            // Perform forwardReaderError, given reader.
336            self.forward_reader_error(cx, self.reader.clone());
337
338            // Retry the pull using the BYOB reader we just acquired.
339            self.pull_with_byob_reader(cx, view, for_branch2, global);
340        }
341    }
342
343    /// Let pullAlgorithm be the following steps:
344    pub(crate) fn pull_algorithm(
345        &self,
346        cx: &mut JSContext,
347        byte_tee_pull_algorithm: Option<ByteTeePullAlgorithm>,
348    ) -> Rc<Promise> {
349        let pull_algorithm =
350            byte_tee_pull_algorithm.unwrap_or(self.byte_tee_pull_algorithm.clone());
351
352        match pull_algorithm {
353            ByteTeePullAlgorithm::Pull1Algorithm => {
354                // If reading is true,
355                if self.reading.get() {
356                    // Set readAgainForBranch1 to true.
357                    self.read_again_for_branch_1.set(true);
358                    // Return a promise resolved with undefined.
359                    return Promise::new_resolved(cx, &self.stream.global(), ());
360                }
361
362                // Set reading to true.
363                self.reading.set(true);
364
365                // Let byobRequest be ! ReadableByteStreamControllerGetBYOBRequest(branch1.[[controller]]).
366                let byob_branch_controller = self
367                    .branch_1
368                    .get()
369                    .expect("Branch 1 should be set.")
370                    .get_byte_controller();
371                let byob_request = byob_branch_controller
372                    .get_byob_request(cx)
373                    .expect("Byob request should be set.");
374
375                match byob_request {
376                    // If byobRequest is null, perform pullWithDefaultReader.
377                    None => {
378                        self.pull_with_default_reader(cx, &self.stream.global())
379                            .expect("Pull with default reader should be successful.");
380                    },
381                    Some(request) => {
382                        // Otherwise, perform pullWithBYOBReader, given byobRequest.[[view]] and false.
383                        let view = request.get_view();
384
385                        self.pull_with_byob_reader(cx, &view, false, &self.stream.global());
386                    },
387                }
388
389                // Return a promise resolved with undefined.
390                Promise::new_resolved(cx, &self.stream.global(), ())
391            },
392            ByteTeePullAlgorithm::Pull2Algorithm => {
393                // If reading is true,
394                if self.reading.get() {
395                    // Set readAgainForBranch2 to true.
396                    self.read_again_for_branch_2.set(true);
397
398                    // Return a promise resolved with undefined.
399                    return Promise::new_resolved(cx, &self.stream.global(), ());
400                }
401
402                // Set reading to true.
403                self.reading.set(true);
404
405                // Let byobRequest be ! ReadableByteStreamControllerGetBYOBRequest(branch2.[[controller]]).
406                let byob_branch_controller = self
407                    .branch_2
408                    .get()
409                    .expect("Branch 2 should be set.")
410                    .get_byte_controller();
411                let byob_request = byob_branch_controller
412                    .get_byob_request(cx)
413                    .expect("Byob request should be set.");
414
415                match byob_request {
416                    None => {
417                        self.pull_with_default_reader(cx, &self.stream.global())
418                            .expect("Pull with default reader should be successful.");
419                    },
420                    Some(request) => {
421                        // Otherwise, perform pullWithBYOBReader, given byobRequest.[[view]] and true.
422                        let view = request.get_view();
423
424                        self.pull_with_byob_reader(cx, &view, true, &self.stream.global());
425                    },
426                }
427
428                // Return a promise resolved with undefined.
429                Promise::new_resolved(cx, &self.stream.global(), ())
430            },
431        }
432    }
433
434    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaulttee>
435    /// Let cancel1Algorithm be the following steps, taking a reason argument
436    /// and
437    /// Let cancel2Algorithm be the following steps, taking a reason argument
438    pub(crate) fn cancel_algorithm(
439        &self,
440        cx: &mut JSContext,
441        reason: SafeHandleValue,
442    ) -> Option<Result<Rc<Promise>, Error>> {
443        match self.tee_cancel_algorithm {
444            ByteTeeCancelAlgorithm::Cancel1Algorithm => {
445                // Set canceled1 to true.
446                self.canceled_1.set(true);
447
448                // Set reason1 to reason.
449                self.reason_1.set(reason.get());
450
451                // If canceled2 is true,
452                if self.canceled_2.get() {
453                    self.resolve_cancel_promise(cx);
454                }
455
456                // Return cancelPromise.
457                Some(Ok(self.cancel_promise.clone()))
458            },
459            ByteTeeCancelAlgorithm::Cancel2Algorithm => {
460                // Set canceled_2 to true.
461                self.canceled_2.set(true);
462
463                // Set reason_2 to reason.
464                self.reason_2.set(reason.get());
465
466                // If canceled_1 is true,
467                if self.canceled_1.get() {
468                    self.resolve_cancel_promise(cx);
469                }
470                // Return cancelPromise.
471                Some(Ok(self.cancel_promise.clone()))
472            },
473        }
474    }
475
476    #[expect(unsafe_code)]
477    fn resolve_cancel_promise(&self, cx: &mut JSContext) {
478        // Let compositeReason be ! CreateArrayFromList(« reason_1, reason_2 »).
479        rooted_vec!(let mut reasons_values);
480        reasons_values.push(self.reason_1.get());
481        reasons_values.push(self.reason_2.get());
482
483        let reasons_values_array = HandleValueArray::from(&reasons_values);
484        rooted!(&in(cx) let reasons = unsafe { NewArrayObject(cx.raw_cx(), &reasons_values_array) });
485        rooted!(&in(cx) let reasons_value = ObjectValue(reasons.get()));
486
487        // Let cancelResult be ! ReadableStreamCancel(stream, compositeReason).
488        let cancel_result = self
489            .stream
490            .cancel(cx, &self.stream.global(), reasons_value.handle());
491
492        // Resolve cancelPromise with cancelResult.
493        self.cancel_promise.resolve_native(cx, &cancel_result);
494    }
495}
script/dom/stream/byteteeunderlyingsource.rs

script/dom/stream/
byteteeunderlyingsource.rs
