Skip to main content

script/dom/stream/
readablebytestreamcontroller.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::cmp::min;
7use std::collections::VecDeque;
8
9use dom_struct::dom_struct;
10use js::context::JSContext;
11use js::jsapi::{Heap, Type};
12use js::jsval::UndefinedValue;
13use js::realm::CurrentRealm;
14use js::rust::{HandleObject, HandleValue as SafeHandleValue, HandleValue};
15use js::typedarray::{ArrayBufferU8, ArrayBufferViewU8};
16use script_bindings::cell::DomRefCell;
17use script_bindings::reflector::{Reflector, reflect_dom_object};
18
19use super::readablestreambyobreader::ReadIntoRequest;
20use super::readablestreamdefaultreader::ReadRequest;
21use super::underlyingsourcecontainer::{UnderlyingSourceContainer, UnderlyingSourceType};
22use crate::dom::bindings::buffer_source::{
23    Constructor, HeapBufferSource, byte_size, create_array_buffer_with_size,
24    create_buffer_source_with_constructor,
25};
26use crate::dom::bindings::codegen::Bindings::ReadableByteStreamControllerBinding::ReadableByteStreamControllerMethods;
27use crate::dom::bindings::codegen::UnionTypes::ReadableStreamDefaultControllerOrReadableByteStreamController as Controller;
28use crate::dom::bindings::error::{Error, ErrorToJsval, Fallible};
29use crate::dom::bindings::reflector::DomGlobal;
30use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
31use crate::dom::bindings::trace::RootedTraceableBox;
32use crate::dom::globalscope::GlobalScope;
33use crate::dom::promise::{Promise, RootedPromise};
34use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
35use crate::dom::stream::readablestream::ReadableStream;
36use crate::dom::stream::readablestreambyobrequest::ReadableStreamBYOBRequest;
37use crate::realms::enter_auto_realm;
38
39/// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry>
40#[derive(JSTraceable, MallocSizeOf)]
41#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
42pub(crate) struct QueueEntry {
43    /// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry-buffer>
44    #[ignore_malloc_size_of = "HeapBufferSource"]
45    buffer: HeapBufferSource<ArrayBufferU8>,
46    /// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry-byte-offset>
47    byte_offset: usize,
48    /// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry-byte-length>
49    byte_length: usize,
50}
51
52impl js::gc::Rootable for QueueEntry {}
53
54impl QueueEntry {
55    pub(crate) fn new(
56        buffer: RootedTraceableBox<HeapBufferSource<ArrayBufferU8>>,
57        byte_offset: usize,
58        byte_length: usize,
59    ) -> QueueEntry {
60        QueueEntry {
61            buffer: *buffer.into_box(),
62            byte_offset,
63            byte_length,
64        }
65    }
66}
67
68#[derive(Debug, Eq, JSTraceable, MallocSizeOf, PartialEq)]
69pub(crate) enum ReaderType {
70    /// <https://streams.spec.whatwg.org/#readablestreambyobreader>
71    Byob,
72    /// <https://streams.spec.whatwg.org/#readablestreamdefaultreader>
73    Default,
74}
75
76/// <https://streams.spec.whatwg.org/#pull-into-descriptor>
77#[derive(Eq, JSTraceable, MallocSizeOf, PartialEq)]
78#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
79pub(crate) struct PullIntoDescriptor {
80    #[ignore_malloc_size_of = "HeapBufferSource"]
81    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-buffer>
82    buffer: HeapBufferSource<ArrayBufferU8>,
83    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-buffer-byte-length>
84    buffer_byte_length: u64,
85    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-byte-offset>
86    byte_offset: u64,
87    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-byte-length>
88    byte_length: u64,
89    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-bytes-filled>
90    bytes_filled: Cell<u64>,
91    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-minimum-fill>
92    minimum_fill: u64,
93    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-element-size>
94    element_size: u64,
95    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-view-constructor>
96    view_constructor: Constructor,
97    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-reader-type>
98    reader_type: Option<ReaderType>,
99}
100
101impl js::gc::Rootable for PullIntoDescriptor {}
102
103/// The fulfillment handler for
104/// <https://streams.spec.whatwg.org/#dom-underlyingsource-start>
105#[derive(Clone, JSTraceable, MallocSizeOf)]
106#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
107struct StartAlgorithmFulfillmentHandler {
108    controller: Dom<ReadableByteStreamController>,
109}
110
111impl Callback for StartAlgorithmFulfillmentHandler {
112    /// Continuation of <https://streams.spec.whatwg.org/#set-up-readable-byte-stream-controller>
113    /// Upon fulfillment of startPromise,
114    fn callback(&self, cx: &mut CurrentRealm, _v: HandleValue) {
115        // Set controller.[[started]] to true.
116        self.controller.started.set(true);
117
118        // Assert: controller.[[pulling]] is false.
119        assert!(!self.controller.pulling.get());
120
121        // Assert: controller.[[pullAgain]] is false.
122        assert!(!self.controller.pull_again.get());
123
124        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
125        self.controller.call_pull_if_needed(cx);
126    }
127}
128
129/// The rejection handler for
130/// <https://streams.spec.whatwg.org/#dom-underlyingsource-start>
131#[derive(Clone, JSTraceable, MallocSizeOf)]
132#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
133struct StartAlgorithmRejectionHandler {
134    controller: Dom<ReadableByteStreamController>,
135}
136
137impl Callback for StartAlgorithmRejectionHandler {
138    /// Continuation of <https://streams.spec.whatwg.org/#set-up-readable-byte-stream-controller>
139    /// Upon rejection of startPromise with reason r,
140    fn callback(&self, cx: &mut CurrentRealm, v: HandleValue) {
141        // Perform ! ReadableByteStreamControllerError(controller, r).
142        self.controller.error(cx, v);
143    }
144}
145
146/// The fulfillment handler for
147/// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
148#[derive(Clone, JSTraceable, MallocSizeOf)]
149#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
150struct PullAlgorithmFulfillmentHandler {
151    controller: Dom<ReadableByteStreamController>,
152}
153
154impl Callback for PullAlgorithmFulfillmentHandler {
155    /// Continuation of <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
156    /// Upon fulfillment of pullPromise
157    fn callback(&self, cx: &mut CurrentRealm, _v: HandleValue) {
158        // Set controller.[[pulling]] to false.
159        self.controller.pulling.set(false);
160
161        // If controller.[[pullAgain]] is true,
162        if self.controller.pull_again.get() {
163            // Set controller.[[pullAgain]] to false.
164            self.controller.pull_again.set(false);
165
166            // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
167            self.controller.call_pull_if_needed(cx);
168        }
169    }
170}
171
172/// The rejection handler for
173/// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
174#[derive(Clone, JSTraceable, MallocSizeOf)]
175#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
176struct PullAlgorithmRejectionHandler {
177    controller: Dom<ReadableByteStreamController>,
178}
179
180impl Callback for PullAlgorithmRejectionHandler {
181    /// Continuation of <https://streams.spec.whatwg.org/#readable-stream-byte-controller-call-pull-if-needed>
182    /// Upon rejection of pullPromise with reason e.
183    fn callback(&self, cx: &mut CurrentRealm, v: HandleValue) {
184        // Perform ! ReadableByteStreamControllerError(controller, e).
185        self.controller.error(cx, v);
186    }
187}
188
189/// <https://streams.spec.whatwg.org/#readablebytestreamcontroller>
190#[dom_struct]
191pub(crate) struct ReadableByteStreamController {
192    reflector_: Reflector,
193    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-autoallocatechunksize>
194    auto_allocate_chunk_size: Option<u64>,
195    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-stream>
196    stream: MutNullableDom<ReadableStream>,
197    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-strategyhwm>
198    strategy_hwm: f64,
199    /// A mutable reference to the underlying source is used to implement these two
200    /// internal slots:
201    ///
202    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pullalgorithm>
203    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-cancelalgorithm>
204    underlying_source: MutNullableDom<UnderlyingSourceContainer>,
205    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-queue>
206    queue: DomRefCell<VecDeque<QueueEntry>>,
207    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-queuetotalsize>
208    queue_total_size: Cell<f64>,
209    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-byobrequest>
210    byob_request: MutNullableDom<ReadableStreamBYOBRequest>,
211    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pendingpullintos>
212    pending_pull_intos: DomRefCell<Vec<PullIntoDescriptor>>,
213    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-closerequested>
214    close_requested: Cell<bool>,
215    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-started>
216    started: Cell<bool>,
217    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pulling>
218    pulling: Cell<bool>,
219    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pullalgorithm>
220    pull_again: Cell<bool>,
221}
222
223impl ReadableByteStreamController {
224    fn new_inherited(
225        underlying_source_container: &UnderlyingSourceContainer,
226        strategy_hwm: f64,
227    ) -> ReadableByteStreamController {
228        let auto_allocate_chunk_size = underlying_source_container.auto_allocate_chunk_size();
229        ReadableByteStreamController {
230            reflector_: Reflector::new(),
231            byob_request: MutNullableDom::new(None),
232            stream: MutNullableDom::new(None),
233            underlying_source: MutNullableDom::new(Some(underlying_source_container)),
234            auto_allocate_chunk_size,
235            pending_pull_intos: DomRefCell::new(Vec::new()),
236            strategy_hwm,
237            close_requested: Default::default(),
238            queue: DomRefCell::new(Default::default()),
239            queue_total_size: Default::default(),
240            started: Default::default(),
241            pulling: Default::default(),
242            pull_again: Default::default(),
243        }
244    }
245
246    pub(crate) fn new(
247        cx: &mut JSContext,
248        underlying_source_type: UnderlyingSourceType,
249        strategy_hwm: f64,
250        global: &GlobalScope,
251    ) -> DomRoot<ReadableByteStreamController> {
252        let underlying_source_container =
253            UnderlyingSourceContainer::new(cx, global, underlying_source_type);
254        reflect_dom_object(
255            cx,
256            Box::new(ReadableByteStreamController::new_inherited(
257                &underlying_source_container,
258                strategy_hwm,
259            )),
260            global,
261        )
262    }
263
264    #[allow(dead_code)]
265    pub(crate) fn set_stream(&self, stream: &ReadableStream) {
266        self.stream.set(Some(stream))
267    }
268
269    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-pull-into>
270    pub(crate) fn perform_pull_into(
271        &self,
272        cx: &mut JSContext,
273        read_into_request: &ReadIntoRequest,
274        view: &HeapBufferSource<ArrayBufferViewU8>,
275        min: u64,
276    ) {
277        // Let stream be controller.[[stream]].
278        let stream = self.stream.get().unwrap();
279
280        // Let elementSize be 1.
281        let mut element_size = 1;
282
283        // Let ctor be %DataView%.
284        let mut ctor = Constructor::DataView;
285
286        // If view has a [[TypedArrayName]] internal slot (i.e., it is not a DataView),
287        if view.has_typed_array_name() {
288            // Set elementSize to the element size specified in the
289            // typed array constructors table for view.[[TypedArrayName]].
290            let view_typw = view.get_array_buffer_view_type();
291            element_size = byte_size(view_typw);
292
293            // Set ctor to the constructor specified in the typed array constructors table for view.[[TypedArrayName]].
294            ctor = Constructor::Name(view_typw);
295        }
296
297        // Let minimumFill be min × elementSize.
298        let minimum_fill = min * element_size;
299
300        // Assert: minimumFill ≥ 0 and minimumFill ≤ view.[[ByteLength]].
301        assert!(minimum_fill <= (view.byte_length() as u64));
302
303        // Assert: the remainder after dividing minimumFill by elementSize is 0.
304        assert_eq!(minimum_fill % element_size, 0);
305
306        // Let byteOffset be view.[[ByteOffset]].
307        let byte_offset = view.get_byte_offset();
308
309        // Let byteLength be view.[[ByteLength]].
310        let byte_length = view.byte_length();
311
312        // Let bufferResult be TransferArrayBuffer(view.[[ViewedArrayBuffer]]).
313        match view
314            .get_array_buffer_view_buffer(cx)
315            .transfer_array_buffer(cx)
316        {
317            Ok(buffer) => {
318                // Let buffer be bufferResult.[[Value]].
319                // Let pullIntoDescriptor be a new pull-into descriptor with
320                // buffer   buffer
321                // buffer byte length   buffer.[[ArrayBufferByteLength]]
322                // byte offset  byteOffset
323                // byte length  byteLength
324                // bytes filled  0
325                // minimum fill minimumFill
326                // element size elementSize
327                // view constructor ctor
328                // reader type  "byob"
329                let buffer_byte_length = buffer.byte_length();
330                let pull_into_descriptor = RootedTraceableBox::new(PullIntoDescriptor {
331                    buffer: *buffer.into_box(),
332                    buffer_byte_length: buffer_byte_length as u64,
333                    byte_offset: byte_offset as u64,
334                    byte_length: byte_length as u64,
335                    bytes_filled: Cell::new(0),
336                    minimum_fill,
337                    element_size,
338                    view_constructor: ctor.clone(),
339                    reader_type: Some(ReaderType::Byob),
340                });
341
342                // If controller.[[pendingPullIntos]] is not empty,
343                {
344                    let mut pending_pull_intos = self.pending_pull_intos.safe_borrow_mut(cx);
345                    if !pending_pull_intos.is_empty() {
346                        // Append pullIntoDescriptor to controller.[[pendingPullIntos]].
347                        pending_pull_intos.push(*pull_into_descriptor.into_box());
348
349                        // Perform ! ReadableStreamAddReadIntoRequest(stream, readIntoRequest).
350                        stream.add_read_into_request(read_into_request);
351
352                        // Return.
353                        return;
354                    }
355                }
356
357                // If stream.[[state]] is "closed",
358                if stream.is_closed() {
359                    // Let emptyView be ! Construct(ctor, « pullIntoDescriptor’s buffer,
360                    // pullIntoDescriptor’s byte offset, 0 »).
361                    if let Ok(empty_view) = create_buffer_source_with_constructor(
362                        cx,
363                        &ctor,
364                        &pull_into_descriptor.buffer,
365                        pull_into_descriptor.byte_offset as usize,
366                        0,
367                    ) {
368                        // Perform readIntoRequest’s close steps, given emptyView.
369                        let result = RootedTraceableBox::new(Heap::default());
370                        rooted!(&in(cx) let mut view_value = UndefinedValue());
371                        empty_view.get_buffer_view_value(cx, view_value.handle_mut());
372                        result.set(*view_value);
373
374                        read_into_request.close_steps(cx, Some(result));
375
376                        // Return.
377                        return;
378                    } else {
379                        return;
380                    }
381                }
382
383                // If controller.[[queueTotalSize]] > 0,
384                if self.queue_total_size.get() > 0.0 {
385                    // If ! ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(
386                    // controller, pullIntoDescriptor) is true,
387                    if self.fill_pull_into_descriptor_from_queue(cx, &pull_into_descriptor) {
388                        // Let filledView be ! ReadableByteStreamControllerConvertPullIntoDescriptor(
389                        // pullIntoDescriptor).
390                        if let Ok(filled_view) =
391                            self.convert_pull_into_descriptor(cx, &pull_into_descriptor)
392                        {
393                            // Perform ! ReadableByteStreamControllerHandleQueueDrain(controller).
394                            self.handle_queue_drain(cx);
395
396                            // Perform readIntoRequest’s chunk steps, given filledView.
397                            let result = RootedTraceableBox::new(Heap::default());
398                            rooted!(&in(cx) let mut view_value = UndefinedValue());
399                            filled_view.get_buffer_view_value(cx, view_value.handle_mut());
400                            result.set(*view_value);
401                            read_into_request.chunk_steps(cx, result);
402
403                            // Return.
404                            return;
405                        } else {
406                            return;
407                        }
408                    }
409
410                    // If controller.[[closeRequested]] is true,
411                    if self.close_requested.get() {
412                        // Let e be a new TypeError exception.
413                        rooted!(&in(cx) let mut error = UndefinedValue());
414                        Error::Type(c"close requested".to_owned()).to_jsval(
415                            cx,
416                            &self.global(),
417                            error.handle_mut(),
418                        );
419
420                        // Perform ! ReadableByteStreamControllerError(controller, e).
421                        self.error(cx, error.handle());
422
423                        // Perform readIntoRequest’s error steps, given e.
424                        read_into_request.error_steps(cx, error.handle());
425
426                        // Return.
427                        return;
428                    }
429                }
430
431                // Append pullIntoDescriptor to controller.[[pendingPullIntos]].
432                {
433                    self.pending_pull_intos
434                        .safe_borrow_mut(cx)
435                        .push(*pull_into_descriptor.into_box());
436                }
437                // Perform ! ReadableStreamAddReadIntoRequest(stream, readIntoRequest).
438                stream.add_read_into_request(read_into_request);
439
440                // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
441                self.call_pull_if_needed(cx);
442            },
443            Err(error) => {
444                // If bufferResult is an abrupt completion,
445
446                // Perform readIntoRequest’s error steps, given bufferResult.[[Value]].
447                rooted!(&in(cx) let mut rval = UndefinedValue());
448                error.to_jsval(cx, &self.global(), rval.handle_mut());
449                read_into_request.error_steps(cx, rval.handle());
450
451                // Return.
452            },
453        }
454    }
455
456    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond>
457    pub(crate) fn respond(&self, cx: &mut JSContext, bytes_written: u64) -> Fallible<()> {
458        let heap_buffer = {
459            // Assert: controller.[[pendingPullIntos]] is not empty.
460            let pending_pull_intos = self.pending_pull_intos.borrow();
461            assert!(!pending_pull_intos.is_empty());
462
463            // Let firstDescriptor be controller.[[pendingPullIntos]][0].
464            let first_descriptor = pending_pull_intos.first().unwrap();
465
466            // Let state be controller.[[stream]].[[state]].
467            let stream = self.stream.get().unwrap();
468
469            // If state is "closed",
470            if stream.is_closed() {
471                // If bytesWritten is not 0, throw a TypeError exception.
472                if bytes_written != 0 {
473                    return Err(Error::Type(
474                        c"bytesWritten not zero on closed stream".to_owned(),
475                    ));
476                }
477            } else {
478                // Assert: state is "readable".
479                assert!(stream.is_readable());
480
481                // If bytesWritten is 0, throw a TypeError exception.
482                if bytes_written == 0 {
483                    return Err(Error::Type(c"bytesWritten is 0".to_owned()));
484                }
485
486                // If firstDescriptor’s bytes filled + bytesWritten > firstDescriptor’s byte length,
487                // throw a RangeError exception.
488                if first_descriptor.bytes_filled.get() + bytes_written >
489                    first_descriptor.byte_length
490                {
491                    return Err(Error::Range(
492                        c"bytes filled + bytesWritten > byte length".to_owned(),
493                    ));
494                }
495            }
496
497            first_descriptor
498                .buffer
499                .transfer_array_buffer(cx)
500                .expect("TransferArrayBuffer failed")
501        };
502        // Set firstDescriptor’s buffer to ! TransferArrayBuffer(firstDescriptor’s buffer).
503        self.pending_pull_intos
504            .safe_borrow_mut(cx)
505            .first_mut()
506            .unwrap()
507            .buffer = *(heap_buffer.into_box());
508
509        // Perform ? ReadableByteStreamControllerRespondInternal(controller, bytesWritten).
510        self.respond_internal(cx, bytes_written)
511    }
512
513    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-internal>
514    fn respond_internal(&self, cx: &mut JSContext, bytes_written: u64) -> Fallible<()> {
515        {
516            // Let firstDescriptor be controller.[[pendingPullIntos]][0].
517            let pending_pull_intos = self.pending_pull_intos.borrow();
518            let first_descriptor = pending_pull_intos.first().unwrap();
519
520            // Assert: ! CanTransferArrayBuffer(firstDescriptor’s buffer) is true
521            assert!(first_descriptor.buffer.can_transfer_array_buffer(cx));
522        }
523
524        // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
525        self.invalidate_byob_request();
526
527        // Let state be controller.[[stream]].[[state]].
528        let stream = self.stream.get().unwrap();
529
530        // If state is "closed",
531        if stream.is_closed() {
532            // Assert: bytesWritten is 0.
533            assert_eq!(bytes_written, 0);
534
535            // Perform ! ReadableByteStreamControllerRespondInClosedState(controller, firstDescriptor).
536            self.respond_in_closed_state(cx)
537                .expect("respond_in_closed_state failed");
538        } else {
539            // Assert: state is "readable".
540            assert!(stream.is_readable());
541
542            // Assert: bytesWritten > 0.
543            assert!(bytes_written > 0);
544
545            // Perform ? ReadableByteStreamControllerRespondInReadableState(controller, bytesWritten, firstDescriptor).
546            self.respond_in_readable_state(cx, bytes_written)?;
547        }
548
549        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
550        self.call_pull_if_needed(cx);
551
552        Ok(())
553    }
554
555    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-in-closed-state>
556    fn respond_in_closed_state(&self, cx: &mut JSContext) -> Fallible<()> {
557        let pending_pull_intos = self.pending_pull_intos.borrow();
558        let first_descriptor = pending_pull_intos.first().unwrap();
559
560        // Assert: the remainder after dividing firstDescriptor’s bytes filled
561        // by firstDescriptor’s element size is 0.
562        assert_eq!(
563            first_descriptor.bytes_filled.get() % first_descriptor.element_size,
564            0
565        );
566
567        // If firstDescriptor’s reader type is "none",
568        // perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
569        let reader_type = first_descriptor.reader_type.is_none();
570
571        // needed to drop the borrow and avoid BorrowMutError
572        drop(pending_pull_intos);
573
574        if reader_type {
575            self.shift_pending_pull_into();
576        }
577
578        // Let stream be controller.[[stream]].
579        let stream = self.stream.get().unwrap();
580
581        // If ! ReadableStreamHasBYOBReader(stream) is true,
582        if stream.has_byob_reader() {
583            // Let filledPullIntos be a new empty list.
584            rooted!(&in(cx) let mut filled_pull_intos = Vec::new());
585
586            // While filledPullIntos’s size < ! ReadableStreamGetNumReadIntoRequests(stream),
587            while filled_pull_intos.len() < stream.get_num_read_into_requests() {
588                // Let pullIntoDescriptor be ! ReadableByteStreamControllerShiftPendingPullInto(controller).
589                // Append pullIntoDescriptor to filledPullIntos.
590                filled_pull_intos.push(self.shift_pending_pull_into());
591            }
592
593            // For each filledPullInto of filledPullIntos,
594            for filled_pull_into in &*filled_pull_intos {
595                // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(stream, filledPullInto).
596                self.commit_pull_into_descriptor(cx, filled_pull_into)
597                    .expect("commit_pull_into_descriptor failed");
598            }
599        }
600
601        Ok(())
602    }
603
604    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-in-readable-state>
605    fn respond_in_readable_state(&self, cx: &mut JSContext, bytes_written: u64) -> Fallible<()> {
606        let pending_pull_intos = self.pending_pull_intos.borrow();
607        let first_descriptor = pending_pull_intos.first().unwrap();
608
609        // Assert: pullIntoDescriptor’s bytes filled + bytesWritten ≤ pullIntoDescriptor’s byte length.
610        assert!(
611            first_descriptor.bytes_filled.get() + bytes_written <= first_descriptor.byte_length
612        );
613
614        // Perform ! ReadableByteStreamControllerFillHeadPullIntoDescriptor(
615        // controller, bytesWritten, pullIntoDescriptor).
616        self.fill_head_pull_into_descriptor(bytes_written, first_descriptor);
617
618        // If pullIntoDescriptor’s reader type is "none",
619        if first_descriptor.reader_type.is_none() {
620            // needed to drop the borrow and avoid BorrowMutError
621            drop(pending_pull_intos);
622
623            // Perform ? ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue(controller, pullIntoDescriptor).
624            self.enqueue_detached_pull_into_to_queue(cx)?;
625
626            // Let filledPullIntos be the result of performing
627            // ! ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
628            rooted!(&in(cx) let filled_pull_intos = self.process_pull_into_descriptors_using_queue(cx));
629
630            // For each filledPullInto of filledPullIntos,
631            for filled_pull_into in &*filled_pull_intos {
632                // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]]
633                // , filledPullInto).
634                self.commit_pull_into_descriptor(cx, filled_pull_into)
635                    .expect("commit_pull_into_descriptor failed");
636            }
637
638            // Return.
639            return Ok(());
640        }
641
642        // If pullIntoDescriptor’s bytes filled < pullIntoDescriptor’s minimum fill, return.
643        if first_descriptor.bytes_filled.get() < first_descriptor.minimum_fill {
644            return Ok(());
645        }
646
647        // needed to drop the borrow and avoid BorrowMutError
648        drop(pending_pull_intos);
649
650        // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
651        rooted!(&in(cx) let pull_into_descriptor = self.shift_pending_pull_into());
652
653        // Let remainderSize be the remainder after dividing pullIntoDescriptor’s bytes
654        // filled by pullIntoDescriptor’s element size.
655        let remainder_size =
656            pull_into_descriptor.bytes_filled.get() % pull_into_descriptor.element_size;
657
658        // If remainderSize > 0,
659        if remainder_size > 0 {
660            // Let end be pullIntoDescriptor’s byte offset + pullIntoDescriptor’s bytes filled.
661            let end = pull_into_descriptor.byte_offset + pull_into_descriptor.bytes_filled.get();
662
663            // Perform ? ReadableByteStreamControllerEnqueueClonedChunkToQueue(controller,
664            // pullIntoDescriptor’s buffer, end − remainderSize, remainderSize).
665            self.enqueue_cloned_chunk_to_queue(
666                cx,
667                &pull_into_descriptor.buffer,
668                end - remainder_size,
669                remainder_size,
670            )?;
671        }
672
673        // Set pullIntoDescriptor’s bytes filled to pullIntoDescriptor’s bytes filled − remainderSize.
674        pull_into_descriptor
675            .bytes_filled
676            .set(pull_into_descriptor.bytes_filled.get() - remainder_size);
677
678        // Let filledPullIntos be the result of performing
679        // ! ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
680        rooted!(&in(cx) let filled_pull_intos = self.process_pull_into_descriptors_using_queue(cx));
681
682        // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]], pullIntoDescriptor).
683        self.commit_pull_into_descriptor(cx, &pull_into_descriptor)
684            .expect("commit_pull_into_descriptor failed");
685
686        // For each filledPullInto of filledPullIntos,
687        for filled_pull_into in &*filled_pull_intos {
688            // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]], filledPullInto).
689            self.commit_pull_into_descriptor(cx, filled_pull_into)
690                .expect("commit_pull_into_descriptor failed");
691        }
692
693        Ok(())
694    }
695
696    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-with-new-view>
697    pub(crate) fn respond_with_new_view(
698        &self,
699        cx: &mut JSContext,
700        view: &HeapBufferSource<ArrayBufferViewU8>,
701    ) -> Fallible<()> {
702        let view_byte_length;
703
704        let view = {
705            // Assert: controller.[[pendingPullIntos]] is not empty.
706            let pending_pull_intos = self.pending_pull_intos.borrow();
707            assert!(!pending_pull_intos.is_empty());
708
709            // Assert: ! IsDetachedBuffer(view.[[ViewedArrayBuffer]]) is false.
710            assert!(!view.is_detached_buffer(cx));
711
712            // Let firstDescriptor be controller.[[pendingPullIntos]][0].
713            let first_descriptor = pending_pull_intos.first().unwrap();
714
715            // Let state be controller.[[stream]].[[state]].
716            let stream = self.stream.get().unwrap();
717
718            // If state is "closed",
719            if stream.is_closed() {
720                // If view.[[ByteLength]] is not 0, throw a TypeError exception.
721                if view.byte_length() != 0 {
722                    return Err(Error::Type(c"view byte length is not 0".to_owned()));
723                }
724            } else {
725                // Assert: state is "readable".
726                assert!(stream.is_readable());
727
728                // If view.[[ByteLength]] is 0, throw a TypeError exception.
729                if view.byte_length() == 0 {
730                    return Err(Error::Type(c"view byte length is 0".to_owned()));
731                }
732            }
733
734            // If firstDescriptor’s byte offset + firstDescriptor’ bytes filled is not view.[[ByteOffset]],
735            // throw a RangeError exception.
736            if first_descriptor.byte_offset + first_descriptor.bytes_filled.get() !=
737                (view.get_byte_offset() as u64)
738            {
739                return Err(Error::Range(
740                    c"firstDescriptor's byte offset + bytes filled is not view byte offset"
741                        .to_owned(),
742                ));
743            }
744
745            // If firstDescriptor’s buffer byte length is not view.[[ViewedArrayBuffer]].[[ByteLength]],
746            // throw a RangeError exception.
747            if first_descriptor.buffer_byte_length !=
748                (view.viewed_buffer_array_byte_length(cx) as u64)
749            {
750                return Err(Error::Range(
751                c"firstDescriptor's buffer byte length is not view viewed buffer array byte length"
752                    .to_owned(),
753            ));
754            }
755
756            // If firstDescriptor’s bytes filled + view.[[ByteLength]] > firstDescriptor’s byte length,
757            // throw a RangeError exception.
758            if first_descriptor.bytes_filled.get() + (view.byte_length()) as u64 >
759                first_descriptor.byte_length
760            {
761                return Err(Error::Range(
762                    c"bytes filled + view byte length > byte length".to_owned(),
763                ));
764            }
765
766            // Let viewByteLength be view.[[ByteLength]].
767            view_byte_length = view.byte_length();
768
769            view.get_array_buffer_view_buffer(cx)
770                .transfer_array_buffer(cx)?
771        };
772        // Set firstDescriptor’s buffer to ? TransferArrayBuffer(view.[[ViewedArrayBuffer]]).
773        self.pending_pull_intos
774            .safe_borrow_mut(cx)
775            .first_mut()
776            .unwrap()
777            .buffer = *view.into_box();
778
779        // Perform ? ReadableByteStreamControllerRespondInternal(controller, viewByteLength).
780        self.respond_internal(cx, view_byte_length as u64)
781    }
782
783    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-get-desired-size>
784    pub(crate) fn get_desired_size(&self) -> Option<f64> {
785        // Let state be controller.[[stream]].[[state]].
786        let stream = self.stream.get()?;
787
788        // If state is "errored", return null.
789        if stream.is_errored() {
790            return None;
791        }
792
793        // If state is "closed", return 0.
794        if stream.is_closed() {
795            return Some(0.0);
796        }
797
798        // Return controller.[[strategyHWM]] − controller.[[queueTotalSize]].
799        Some(self.strategy_hwm - self.queue_total_size.get())
800    }
801
802    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollergetbyobrequest>
803    pub(crate) fn get_byob_request(
804        &self,
805        cx: &mut js::context::JSContext,
806    ) -> Fallible<Option<DomRoot<ReadableStreamBYOBRequest>>> {
807        // If controller.[[byobRequest]] is null and controller.[[pendingPullIntos]] is not empty,
808        let pending_pull_intos = self.pending_pull_intos.borrow();
809        if self.byob_request.get().is_none() && !pending_pull_intos.is_empty() {
810            // Let firstDescriptor be controller.[[pendingPullIntos]][0].
811            let first_descriptor = pending_pull_intos.first().unwrap();
812            // Let view be ! Construct(%Uint8Array%, « firstDescriptor’s buffer,
813            // firstDescriptor’s byte offset + firstDescriptor’s bytes filled,
814            // firstDescriptor’s byte length − firstDescriptor’s bytes filled »).
815
816            let byte_offset = first_descriptor.byte_offset + first_descriptor.bytes_filled.get();
817            let byte_length = first_descriptor.byte_length - first_descriptor.bytes_filled.get();
818
819            let view = create_buffer_source_with_constructor(
820                cx,
821                &Constructor::Name(Type::Uint8),
822                &first_descriptor.buffer,
823                byte_offset as usize,
824                byte_length as usize,
825            )
826            .expect("Construct Uint8Array failed");
827
828            // Let byobRequest be a new ReadableStreamBYOBRequest.
829            let byob_request = ReadableStreamBYOBRequest::new(cx, &self.global());
830
831            // Set byobRequest.[[controller]] to controller.
832            byob_request.set_controller(Some(&DomRoot::from_ref(self)));
833
834            // Set byobRequest.[[view]] to view.
835            byob_request.set_view(Some(view));
836
837            // Set controller.[[byobRequest]] to byobRequest.
838            self.byob_request.set(Some(&byob_request));
839        }
840
841        // Return controller.[[byobRequest]].
842        Ok(self.byob_request.get())
843    }
844
845    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-close>
846    pub(crate) fn close(&self, cx: &mut JSContext) -> Fallible<()> {
847        // Let stream be controller.[[stream]].
848        let stream = self.stream.get().unwrap();
849
850        // If controller.[[closeRequested]] is true or stream.[[state]] is not "readable", return.
851        if self.close_requested.get() || !stream.is_readable() {
852            return Ok(());
853        }
854
855        // If controller.[[queueTotalSize]] > 0,
856        if self.queue_total_size.get() > 0.0 {
857            // Set controller.[[closeRequested]] to true.
858            self.close_requested.set(true);
859            // Return.
860            return Ok(());
861        }
862
863        {
864            // If controller.[[pendingPullIntos]] is not empty,
865            let pending_pull_intos = self.pending_pull_intos.borrow();
866            if !pending_pull_intos.is_empty() {
867                // Let firstPendingPullInto be controller.[[pendingPullIntos]][0].
868                let first_pending_pull_into = pending_pull_intos.first().unwrap();
869
870                // If the remainder after dividing firstPendingPullInto’s bytes filled by
871                // firstPendingPullInto’s element size is not 0,
872                if !first_pending_pull_into
873                    .bytes_filled
874                    .get()
875                    .is_multiple_of(first_pending_pull_into.element_size)
876                {
877                    // needed to drop the borrow and avoid BorrowMutError
878                    drop(pending_pull_intos);
879
880                    // Let e be a new TypeError exception.
881                    let e = Error::Type(
882                        c"remainder after dividing firstPendingPullInto's bytes
883                    filled by firstPendingPullInto's element size is not 0"
884                            .to_owned(),
885                    );
886
887                    // Perform ! ReadableByteStreamControllerError(controller, e).
888                    rooted!(&in(cx) let mut error = UndefinedValue());
889                    e.clone().to_jsval(cx, &self.global(), error.handle_mut());
890                    self.error(cx, error.handle());
891
892                    // Throw e.
893                    return Err(e);
894                }
895            }
896        }
897
898        // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
899        self.clear_algorithms();
900
901        // Perform ! ReadableStreamClose(stream).
902        stream.close(cx);
903        Ok(())
904    }
905
906    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-error>
907    pub(crate) fn error(&self, cx: &mut JSContext, e: SafeHandleValue) {
908        // Let stream be controller.[[stream]].
909        let stream = self.stream.get().unwrap();
910
911        // If stream.[[state]] is not "readable", return.
912        if !stream.is_readable() {
913            return;
914        }
915
916        // Perform ! ReadableByteStreamControllerClearPendingPullIntos(controller).
917        self.clear_pending_pull_intos();
918
919        // Perform ! ResetQueue(controller).
920        self.reset_queue();
921
922        // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
923        self.clear_algorithms();
924
925        // Perform ! ReadableStreamError(stream, e).
926        stream.error(cx, e);
927    }
928
929    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-clear-algorithms>
930    fn clear_algorithms(&self) {
931        // Set controller.[[pullAlgorithm]] to undefined.
932        // Set controller.[[cancelAlgorithm]] to undefined.
933        self.underlying_source.set(None);
934    }
935
936    /// <https://streams.spec.whatwg.org/#reset-queue>
937    pub(crate) fn reset_queue(&self) {
938        // Assert: container has [[queue]] and [[queueTotalSize]] internal slots.
939
940        // Set container.[[queue]] to a new empty list.
941        self.queue.borrow_mut().clear();
942
943        // Set container.[[queueTotalSize]] to 0.
944        self.queue_total_size.set(0.0);
945    }
946
947    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-clear-pending-pull-intos>
948    pub(crate) fn clear_pending_pull_intos(&self) {
949        // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
950        self.invalidate_byob_request();
951
952        // Set controller.[[pendingPullIntos]] to a new empty list.
953        self.pending_pull_intos.borrow_mut().clear();
954    }
955
956    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-invalidate-byob-request>
957    pub(crate) fn invalidate_byob_request(&self) {
958        if let Some(byob_request) = self.byob_request.get() {
959            // Set controller.[[byobRequest]].[[controller]] to undefined.
960            byob_request.set_controller(None);
961
962            // Set controller.[[byobRequest]].[[view]] to null.
963            byob_request.set_view(None);
964
965            // Set controller.[[byobRequest]] to null.
966            self.byob_request.set(None);
967        }
968        // If controller.[[byobRequest]] is null, return.
969    }
970
971    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-enqueue>
972    pub(crate) fn enqueue(
973        &self,
974        cx: &mut JSContext,
975        chunk: RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>,
976    ) -> Fallible<()> {
977        // Let stream be controller.[[stream]].
978        let stream = self.stream.get().unwrap();
979
980        // If controller.[[closeRequested]] is true or stream.[[state]] is not "readable", return.
981        if self.close_requested.get() || !stream.is_readable() {
982            return Ok(());
983        }
984
985        // Let buffer be chunk.[[ViewedArrayBuffer]].
986        let buffer = chunk.get_array_buffer_view_buffer(cx);
987
988        // Let byteOffset be chunk.[[ByteOffset]].
989        let byte_offset = chunk.get_byte_offset();
990
991        // Let byteLength be chunk.[[ByteLength]].
992        let byte_length = chunk.byte_length();
993
994        // If ! IsDetachedBuffer(buffer) is true, throw a TypeError exception.
995        if buffer.is_detached_buffer(cx) {
996            return Err(Error::Type(c"buffer is detached".to_owned()));
997        }
998
999        // Let transferredBuffer be ? TransferArrayBuffer(buffer).
1000        let transferred_buffer = buffer.transfer_array_buffer(cx)?;
1001
1002        // If controller.[[pendingPullIntos]] is not empty,
1003
1004        let pending_pull_intos = self.pending_pull_intos.borrow();
1005        if !pending_pull_intos.is_empty() {
1006            let heap_buffer = {
1007                // Let firstPendingPullInto be controller.[[pendingPullIntos]][0].
1008                let first_descriptor = pending_pull_intos.first().unwrap();
1009                // If ! IsDetachedBuffer(firstPendingPullInto’s buffer) is true, throw a TypeError exception.
1010                if first_descriptor.buffer.is_detached_buffer(cx) {
1011                    return Err(Error::Type(c"buffer is detached".to_owned()));
1012                }
1013
1014                // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
1015                self.invalidate_byob_request();
1016
1017                first_descriptor
1018                    .buffer
1019                    .transfer_array_buffer(cx)
1020                    .expect("TransferArrayBuffer failed")
1021            };
1022
1023            drop(pending_pull_intos);
1024            // Set firstPendingPullInto’s buffer to ! TransferArrayBuffer(firstPendingPullInto’s buffer).
1025
1026            self.pending_pull_intos
1027                .safe_borrow_mut(cx)
1028                .first_mut()
1029                .unwrap()
1030                .buffer = *heap_buffer.into_box();
1031
1032            // If firstPendingPullInto’s reader type is "none",
1033            if self
1034                .pending_pull_intos
1035                .borrow()
1036                .first()
1037                .unwrap()
1038                .reader_type
1039                .is_none()
1040            {
1041                // perform ? ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue(
1042                // controller, firstPendingPullInto).
1043                self.enqueue_detached_pull_into_to_queue(cx)?;
1044            }
1045        }
1046
1047        // If ! ReadableStreamHasDefaultReader(stream) is true,
1048        if stream.has_default_reader() {
1049            // Perform ! ReadableByteStreamControllerProcessReadRequestsUsingQueue(controller).
1050            self.process_read_requests_using_queue(cx)
1051                .expect("process_read_requests_using_queue failed");
1052
1053            // If ! ReadableStreamGetNumReadRequests(stream) is 0,
1054            if stream.get_num_read_requests() == 0 {
1055                // Assert: controller.[[pendingPullIntos]] is empty.
1056                {
1057                    assert!(self.pending_pull_intos.borrow().is_empty());
1058                }
1059
1060                // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue(
1061                // controller, transferredBuffer, byteOffset, byteLength).
1062                self.enqueue_chunk_to_queue(transferred_buffer, byte_offset, byte_length);
1063            } else {
1064                // Assert: controller.[[queue]] is empty.
1065                assert!(self.queue.borrow().is_empty());
1066
1067                // If controller.[[pendingPullIntos]] is not empty,
1068
1069                let pending_pull_intos = self.pending_pull_intos.borrow();
1070                if !pending_pull_intos.is_empty() {
1071                    // Assert: controller.[[pendingPullIntos]][0]'s reader type is "default".
1072                    assert!(matches!(
1073                        pending_pull_intos.first().unwrap().reader_type,
1074                        Some(ReaderType::Default)
1075                    ));
1076
1077                    // needed to drop the borrow and avoid BorrowMutError
1078                    drop(pending_pull_intos);
1079
1080                    // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
1081                    self.shift_pending_pull_into();
1082                }
1083
1084                // Let transferredView be ! Construct(%Uint8Array%, « transferredBuffer, byteOffset, byteLength »).
1085                let transferred_view = create_buffer_source_with_constructor(
1086                    cx,
1087                    &Constructor::Name(Type::Uint8),
1088                    &transferred_buffer,
1089                    byte_offset,
1090                    byte_length,
1091                )
1092                .expect("Construct Uint8Array failed");
1093
1094                // Perform ! ReadableStreamFulfillReadRequest(stream, transferredView, false).
1095                rooted!(&in(cx) let mut view_value = UndefinedValue());
1096                transferred_view.get_buffer_view_value(cx, view_value.handle_mut());
1097                stream.fulfill_read_request(cx, view_value.handle(), false);
1098            }
1099            // Otherwise, if ! ReadableStreamHasBYOBReader(stream) is true,
1100        } else if stream.has_byob_reader() {
1101            // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue(
1102            // controller, transferredBuffer, byteOffset, byteLength).
1103            self.enqueue_chunk_to_queue(transferred_buffer, byte_offset, byte_length);
1104
1105            // Let filledPullIntos be the result of performing !
1106            // ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
1107            rooted!(&in(cx) let filled_pull_intos = self.process_pull_into_descriptors_using_queue(cx));
1108
1109            // For each filledPullInto of filledPullIntos,
1110            // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(stream, filledPullInto).
1111            for filled_pull_into in &*filled_pull_intos {
1112                self.commit_pull_into_descriptor(cx, filled_pull_into)
1113                    .expect("commit_pull_into_descriptor failed");
1114            }
1115        } else {
1116            // Assert: ! IsReadableStreamLocked(stream) is false.
1117            assert!(!stream.is_locked());
1118
1119            // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue
1120            // (controller, transferredBuffer, byteOffset, byteLength).
1121            self.enqueue_chunk_to_queue(transferred_buffer, byte_offset, byte_length);
1122        }
1123
1124        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
1125        self.call_pull_if_needed(cx);
1126
1127        Ok(())
1128    }
1129
1130    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-commit-pull-into-descriptor>
1131    fn commit_pull_into_descriptor(
1132        &self,
1133        cx: &mut JSContext,
1134        pull_into_descriptor: &PullIntoDescriptor,
1135    ) -> Fallible<()> {
1136        // Assert: stream.[[state]] is not "errored".
1137        let stream = self.stream.get().unwrap();
1138        assert!(!stream.is_errored());
1139
1140        // Assert: pullIntoDescriptor.reader type is not "none".
1141        assert!(pull_into_descriptor.reader_type.is_some());
1142
1143        // Let done be false.
1144        let mut done = false;
1145
1146        // If stream.[[state]] is "closed",
1147        if stream.is_closed() {
1148            // Assert: the remainder after dividing pullIntoDescriptor’s bytes filled
1149            // by pullIntoDescriptor’s element size is 0.
1150            assert!(
1151                pull_into_descriptor
1152                    .bytes_filled
1153                    .get()
1154                    .is_multiple_of(pull_into_descriptor.element_size)
1155            );
1156
1157            // Set done to true.
1158            done = true;
1159        }
1160
1161        // Let filledView be ! ReadableByteStreamControllerConvertPullIntoDescriptor(pullIntoDescriptor).
1162        let filled_view = self
1163            .convert_pull_into_descriptor(cx, pull_into_descriptor)
1164            .expect("convert_pull_into_descriptor failed");
1165
1166        rooted!(&in(cx) let mut view_value = UndefinedValue());
1167        filled_view.get_buffer_view_value(cx, view_value.handle_mut());
1168
1169        // If pullIntoDescriptor’s reader type is "default",
1170        if matches!(pull_into_descriptor.reader_type, Some(ReaderType::Default)) {
1171            // Perform ! ReadableStreamFulfillReadRequest(stream, filledView, done).
1172
1173            stream.fulfill_read_request(cx, view_value.handle(), done);
1174        } else {
1175            // Assert: pullIntoDescriptor’s reader type is "byob".
1176            assert!(matches!(
1177                pull_into_descriptor.reader_type,
1178                Some(ReaderType::Byob)
1179            ));
1180
1181            // Perform ! ReadableStreamFulfillReadIntoRequest(stream, filledView, done).
1182            stream.fulfill_read_into_request(cx, view_value.handle(), done);
1183        }
1184        Ok(())
1185    }
1186
1187    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-convert-pull-into-descriptor>
1188    pub(crate) fn convert_pull_into_descriptor(
1189        &self,
1190        cx: &mut js::context::JSContext,
1191        pull_into_descriptor: &PullIntoDescriptor,
1192    ) -> Fallible<RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>> {
1193        // Let bytesFilled be pullIntoDescriptor’s bytes filled.
1194        let bytes_filled = pull_into_descriptor.bytes_filled.get();
1195
1196        // Let elementSize be pullIntoDescriptor’s element size.
1197        let element_size = pull_into_descriptor.element_size;
1198
1199        // Assert: bytesFilled ≤ pullIntoDescriptor’s byte length.
1200        assert!(bytes_filled <= pull_into_descriptor.byte_length);
1201
1202        // Assert: the remainder after dividing bytesFilled by elementSize is 0.
1203        assert!(bytes_filled.is_multiple_of(element_size));
1204
1205        // Let buffer be ! TransferArrayBuffer(pullIntoDescriptor’s buffer).
1206        let buffer = pull_into_descriptor
1207            .buffer
1208            .transfer_array_buffer(cx)
1209            .expect("TransferArrayBuffer failed");
1210
1211        // Return ! Construct(pullIntoDescriptor’s view constructor,
1212        // « buffer, pullIntoDescriptor’s byte offset, bytesFilled ÷ elementSize »).
1213        Ok(create_buffer_source_with_constructor(
1214            cx,
1215            &pull_into_descriptor.view_constructor,
1216            &buffer,
1217            pull_into_descriptor.byte_offset as usize,
1218            (bytes_filled / element_size) as usize,
1219        )
1220        .expect("Construct view failed"))
1221    }
1222
1223    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-process-pull-into-descriptors-using-queue>
1224    pub(crate) fn process_pull_into_descriptors_using_queue(
1225        &self,
1226        cx: &mut js::context::JSContext,
1227    ) -> Vec<PullIntoDescriptor> {
1228        // Assert: controller.[[closeRequested]] is false.
1229        assert!(!self.close_requested.get());
1230
1231        // Let filledPullIntos be a new empty list.
1232        rooted!(&in(cx) let mut filled_pull_intos = Vec::new());
1233
1234        // While controller.[[pendingPullIntos]] is not empty,
1235        loop {
1236            // If controller.[[queueTotalSize]] is 0, then break.
1237            if self.queue_total_size.get() == 0.0 {
1238                break;
1239            }
1240
1241            // Let pullIntoDescriptor be controller.[[pendingPullIntos]][0].
1242            let fill_pull_result = {
1243                let pending_pull_intos = self.pending_pull_intos.borrow();
1244                let Some(pull_into_descriptor) = pending_pull_intos.first() else {
1245                    break;
1246                };
1247                self.fill_pull_into_descriptor_from_queue(cx, pull_into_descriptor)
1248            };
1249
1250            // If ! ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller, pullIntoDescriptor) is true,
1251            if fill_pull_result {
1252                // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
1253                // Append pullIntoDescriptor to filledPullIntos.
1254                filled_pull_intos.push(self.shift_pending_pull_into());
1255            }
1256        }
1257
1258        // Return filledPullIntos.
1259        filled_pull_intos.take()
1260    }
1261
1262    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-fill-pull-into-descriptor-from-queue>
1263    pub(crate) fn fill_pull_into_descriptor_from_queue(
1264        &self,
1265        cx: &mut js::context::JSContext,
1266        pull_into_descriptor: &PullIntoDescriptor,
1267    ) -> bool {
1268        // Let maxBytesToCopy be min(controller.[[queueTotalSize]],
1269        // pullIntoDescriptor’s byte length − pullIntoDescriptor’s bytes filled).
1270        let max_bytes_to_copy = min(
1271            self.queue_total_size.get() as usize,
1272            (pull_into_descriptor.byte_length - pull_into_descriptor.bytes_filled.get()) as usize,
1273        );
1274
1275        // Let maxBytesFilled be pullIntoDescriptor’s bytes filled + maxBytesToCopy.
1276        let max_bytes_filled = pull_into_descriptor.bytes_filled.get() as usize + max_bytes_to_copy;
1277
1278        // Let totalBytesToCopyRemaining be maxBytesToCopy.
1279        let mut total_bytes_to_copy_remaining = max_bytes_to_copy;
1280
1281        // Let ready be false.
1282        let mut ready = false;
1283
1284        // Assert: ! IsDetachedBuffer(pullIntoDescriptor’s buffer) is false.
1285        assert!(!pull_into_descriptor.buffer.is_detached_buffer(cx));
1286
1287        // Assert: pullIntoDescriptor’s bytes filled < pullIntoDescriptor’s minimum fill.
1288        assert!(pull_into_descriptor.bytes_filled.get() < pull_into_descriptor.minimum_fill);
1289
1290        // Let remainderBytes be the remainder after dividing maxBytesFilled by pullIntoDescriptor’s element size.
1291        let remainder_bytes = max_bytes_filled % pull_into_descriptor.element_size as usize;
1292
1293        // Let maxAlignedBytes be maxBytesFilled − remainderBytes.
1294        let max_aligned_bytes = max_bytes_filled - remainder_bytes;
1295
1296        // If maxAlignedBytes ≥ pullIntoDescriptor’s minimum fill,
1297        if max_aligned_bytes >= pull_into_descriptor.minimum_fill as usize {
1298            // Set totalBytesToCopyRemaining to maxAlignedBytes − pullIntoDescriptor’s bytes filled.
1299            total_bytes_to_copy_remaining =
1300                max_aligned_bytes - (pull_into_descriptor.bytes_filled.get() as usize);
1301
1302            // Set ready to true.
1303            ready = true;
1304        }
1305
1306        // Let queue be controller.[[queue]].
1307        // While totalBytesToCopyRemaining > 0,
1308        while total_bytes_to_copy_remaining > 0 {
1309            // Let headOfQueue be queue[0].
1310            let queue = self.queue.borrow();
1311            let head_of_queue = queue.front().unwrap();
1312
1313            // Let bytesToCopy be min(totalBytesToCopyRemaining, headOfQueue’s byte length).
1314            let bytes_to_copy = total_bytes_to_copy_remaining.min(head_of_queue.byte_length);
1315
1316            // Let destStart be pullIntoDescriptor’s byte offset + pullIntoDescriptor’s bytes filled.
1317            let dest_start =
1318                pull_into_descriptor.byte_offset + pull_into_descriptor.bytes_filled.get();
1319
1320            // Let descriptorBuffer be pullIntoDescriptor’s buffer.
1321            let descriptor_buffer = &pull_into_descriptor.buffer;
1322
1323            // Let queueBuffer be headOfQueue’s buffer.
1324            let queue_buffer = &head_of_queue.buffer;
1325
1326            // Let queueByteOffset be headOfQueue’s byte offset.
1327            let queue_byte_offset = head_of_queue.byte_offset;
1328
1329            // Assert: ! CanCopyDataBlockBytes(descriptorBuffer, destStart,
1330            // queueBuffer, queueByteOffset, bytesToCopy) is true.
1331            assert!(descriptor_buffer.can_copy_data_block_bytes(
1332                cx,
1333                dest_start as usize,
1334                queue_buffer,
1335                queue_byte_offset,
1336                bytes_to_copy
1337            ));
1338
1339            // Perform ! CopyDataBlockBytes(descriptorBuffer.[[ArrayBufferData]], destStart,
1340            // queueBuffer.[[ArrayBufferData]], queueByteOffset, bytesToCopy).
1341            descriptor_buffer.copy_data_block_bytes(
1342                cx,
1343                dest_start as usize,
1344                queue_buffer,
1345                queue_byte_offset,
1346                bytes_to_copy,
1347            );
1348
1349            let head_of_queue_byte_length = head_of_queue.byte_length;
1350            // Remove the borrow on self.queue
1351            drop(queue);
1352
1353            // If headOfQueue’s byte length is bytesToCopy,
1354            if head_of_queue_byte_length == bytes_to_copy {
1355                // Remove queue[0].
1356                self.queue.safe_borrow_mut(cx).pop_front().unwrap();
1357            } else {
1358                let mut queue = self.queue.safe_borrow_mut(cx);
1359                let head_of_queue = queue.front_mut().unwrap();
1360                // Set headOfQueue’s byte offset to headOfQueue’s byte offset + bytesToCopy.
1361                head_of_queue.byte_offset += bytes_to_copy;
1362
1363                // Set headOfQueue’s byte length to headOfQueue’s byte length − bytesToCopy.
1364                head_of_queue.byte_length -= bytes_to_copy;
1365            }
1366
1367            // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] − bytesToCopy.
1368            self.queue_total_size
1369                .set(self.queue_total_size.get() - (bytes_to_copy as f64));
1370
1371            // Perform ! ReadableByteStreamControllerFillHeadPullIntoDescriptor(
1372            // controller, bytesToCopy, pullIntoDescriptor).
1373            self.fill_head_pull_into_descriptor(bytes_to_copy as u64, pull_into_descriptor);
1374
1375            // Set totalBytesToCopyRemaining to totalBytesToCopyRemaining − bytesToCopy.
1376            total_bytes_to_copy_remaining -= bytes_to_copy;
1377        }
1378
1379        // If ready is false,
1380        if !ready {
1381            // Assert: controller.[[queueTotalSize]] is 0.
1382            assert!(self.queue_total_size.get() == 0.0);
1383
1384            // Assert: pullIntoDescriptor’s bytes filled > 0.
1385            assert!(pull_into_descriptor.bytes_filled.get() > 0);
1386
1387            // Assert: pullIntoDescriptor’s bytes filled < pullIntoDescriptor’s minimum fill.
1388            assert!(pull_into_descriptor.bytes_filled.get() < pull_into_descriptor.minimum_fill);
1389        }
1390
1391        // Return ready.
1392        ready
1393    }
1394
1395    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-fill-head-pull-into-descriptor>
1396    pub(crate) fn fill_head_pull_into_descriptor(
1397        &self,
1398        bytes_copied: u64,
1399        pull_into_descriptor: &PullIntoDescriptor,
1400    ) {
1401        // Assert: either controller.[[pendingPullIntos]] is empty,
1402        // or controller.[[pendingPullIntos]][0] is pullIntoDescriptor.
1403        {
1404            let pending_pull_intos = self.pending_pull_intos.borrow();
1405            assert!(
1406                pending_pull_intos.is_empty() ||
1407                    pending_pull_intos.first().unwrap() == pull_into_descriptor
1408            );
1409        }
1410
1411        // Assert: controller.[[byobRequest]] is null.
1412        assert!(self.byob_request.get().is_none());
1413
1414        // Set pullIntoDescriptor’s bytes filled to bytes filled + size.
1415        pull_into_descriptor
1416            .bytes_filled
1417            .set(pull_into_descriptor.bytes_filled.get() + bytes_copied);
1418    }
1419
1420    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerenqueuedetachedpullintotoqueue>
1421    pub(crate) fn enqueue_detached_pull_into_to_queue(&self, cx: &mut JSContext) -> Fallible<()> {
1422        // first_descriptor: &PullIntoDescriptor,
1423        let pending_pull_intos = self.pending_pull_intos.borrow();
1424        let first_descriptor = pending_pull_intos.first().unwrap();
1425
1426        // Assert: pullIntoDescriptor’s reader type is "none".
1427        assert!(first_descriptor.reader_type.is_none());
1428
1429        // If pullIntoDescriptor’s bytes filled > 0, perform ?
1430        // ReadableByteStreamControllerEnqueueClonedChunkToQueue(controller,
1431        // pullIntoDescriptor’s buffer, pullIntoDescriptor’s byte offset, pullIntoDescriptor’s bytes filled).
1432
1433        if first_descriptor.bytes_filled.get() > 0 {
1434            self.enqueue_cloned_chunk_to_queue(
1435                cx,
1436                &first_descriptor.buffer,
1437                first_descriptor.byte_offset,
1438                first_descriptor.bytes_filled.get(),
1439            )?;
1440        }
1441
1442        // needed to drop the borrow and avoid BorrowMutError
1443        drop(pending_pull_intos);
1444
1445        // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
1446        self.shift_pending_pull_into();
1447
1448        Ok(())
1449    }
1450
1451    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerenqueueclonedchunktoqueue>
1452    pub(crate) fn enqueue_cloned_chunk_to_queue(
1453        &self,
1454        cx: &mut JSContext,
1455        buffer: &HeapBufferSource<ArrayBufferU8>,
1456        byte_offset: u64,
1457        byte_length: u64,
1458    ) -> Fallible<()> {
1459        // Let cloneResult be CloneArrayBuffer(buffer, byteOffset, byteLength, %ArrayBuffer%).
1460        if let Ok(clone_result) =
1461            buffer.clone_array_buffer(cx, byte_offset as usize, byte_length as usize)
1462        {
1463            // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue
1464            // (controller, cloneResult.[[Value]], 0, byteLength).
1465            self.enqueue_chunk_to_queue(clone_result, 0, byte_length as usize);
1466
1467            Ok(())
1468        } else {
1469            // If cloneResult is an abrupt completion,
1470
1471            // Perform ! ReadableByteStreamControllerError(controller, cloneResult.[[Value]]).
1472            rooted!(&in(cx) let mut rval = UndefinedValue());
1473            let error = Error::Type(c"can not clone array buffer".to_owned());
1474            error
1475                .clone()
1476                .to_jsval(cx, &self.global(), rval.handle_mut());
1477            self.error(cx, rval.handle());
1478
1479            // Return cloneResult.
1480            Err(error)
1481        }
1482    }
1483
1484    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-enqueue-chunk-to-queue>
1485    pub(crate) fn enqueue_chunk_to_queue(
1486        &self,
1487        buffer: RootedTraceableBox<HeapBufferSource<ArrayBufferU8>>,
1488        byte_offset: usize,
1489        byte_length: usize,
1490    ) {
1491        // Let entry be a new ReadableByteStreamQueueEntry object.
1492        // Append entry to controller.[[queue]].
1493        self.queue
1494            .borrow_mut()
1495            .push_back(QueueEntry::new(buffer, byte_offset, byte_length));
1496
1497        // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] + byteLength.
1498        self.queue_total_size
1499            .set(self.queue_total_size.get() + byte_length as f64);
1500    }
1501
1502    pub(crate) fn in_memory(&self) -> bool {
1503        let Some(underlying_source) = self.underlying_source.get() else {
1504            return false;
1505        };
1506        underlying_source.in_memory()
1507    }
1508
1509    pub(crate) fn get_in_memory_bytes(&self, cx: &mut JSContext) -> Option<Vec<u8>> {
1510        let underlying_source = self.underlying_source.get()?;
1511        if !underlying_source.in_memory() {
1512            return None;
1513        }
1514
1515        self.queue.borrow().iter().try_fold(
1516            Vec::with_capacity(self.queue_total_size.get() as usize),
1517            |mut bytes, entry| {
1518                let mut chunk = vec![0; entry.byte_length];
1519                entry
1520                    .buffer
1521                    .copy_data_to(
1522                        cx,
1523                        &mut chunk,
1524                        entry.byte_offset,
1525                        entry.byte_offset + entry.byte_length,
1526                    )
1527                    .ok()?;
1528                bytes.extend(chunk);
1529                Some(bytes)
1530            },
1531        )
1532    }
1533
1534    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-shift-pending-pull-into>
1535    pub(crate) fn shift_pending_pull_into(&self) -> PullIntoDescriptor {
1536        // Assert: controller.[[byobRequest]] is null.
1537        assert!(self.byob_request.get().is_none());
1538
1539        // Let descriptor be controller.[[pendingPullIntos]][0].
1540        // Remove descriptor from controller.[[pendingPullIntos]].
1541        // Return descriptor.
1542        self.pending_pull_intos.borrow_mut().remove(0)
1543    }
1544
1545    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerprocessreadrequestsusingqueue>
1546    pub(crate) fn process_read_requests_using_queue(&self, cx: &mut JSContext) -> Fallible<()> {
1547        // Let reader be controller.[[stream]].[[reader]].
1548        // Assert: reader implements ReadableStreamDefaultReader.
1549        let reader = self.stream.get().unwrap().get_default_reader();
1550
1551        // Step 3
1552        reader.process_read_requests(cx, self)
1553    }
1554
1555    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerfillreadrequestfromqueue>
1556    pub(crate) fn fill_read_request_from_queue(
1557        &self,
1558        cx: &mut JSContext,
1559        read_request: &ReadRequest,
1560    ) -> Fallible<()> {
1561        // Assert: controller.[[queueTotalSize]] > 0.
1562        assert!(self.queue_total_size.get() > 0.0);
1563        // Also assert that the queue has a non-zero length;
1564        assert!(!self.queue.borrow().is_empty());
1565
1566        // Let entry be controller.[[queue]][0].
1567        // Remove entry from controller.[[queue]].
1568        rooted!(&in(cx) let entry = self.remove_entry());
1569
1570        // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] − entry’s byte length.
1571        self.queue_total_size
1572            .set(self.queue_total_size.get() - entry.byte_length as f64);
1573
1574        // Perform ! ReadableByteStreamControllerHandleQueueDrain(controller).
1575        self.handle_queue_drain(cx);
1576
1577        // Let view be ! Construct(%Uint8Array%, « entry’s buffer, entry’s byte offset, entry’s byte length »).
1578        let view = create_buffer_source_with_constructor(
1579            cx,
1580            &Constructor::Name(Type::Uint8),
1581            &entry.buffer,
1582            entry.byte_offset,
1583            entry.byte_length,
1584        )
1585        .expect("Construct Uint8Array failed");
1586
1587        // Perform readRequest’s chunk steps, given view.
1588        let result = RootedTraceableBox::new(Heap::default());
1589        rooted!(&in(cx) let mut view_value = UndefinedValue());
1590        view.get_buffer_view_value(cx, view_value.handle_mut());
1591        result.set(*view_value);
1592
1593        read_request.chunk_steps(cx, result, &self.global());
1594
1595        Ok(())
1596    }
1597
1598    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-handle-queue-drain>
1599    pub(crate) fn handle_queue_drain(&self, cx: &mut JSContext) {
1600        // Assert: controller.[[stream]].[[state]] is "readable".
1601        assert!(self.stream.get().unwrap().is_readable());
1602
1603        // If controller.[[queueTotalSize]] is 0 and controller.[[closeRequested]] is true,
1604        if self.queue_total_size.get() == 0.0 && self.close_requested.get() {
1605            // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
1606            self.clear_algorithms();
1607
1608            // Perform ! ReadableStreamClose(controller.[[stream]]).
1609            self.stream.get().unwrap().close(cx);
1610        } else {
1611            // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
1612            self.call_pull_if_needed(cx);
1613        }
1614    }
1615
1616    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
1617    fn call_pull_if_needed(&self, cx: &mut JSContext) {
1618        // Let shouldPull be ! ReadableByteStreamControllerShouldCallPull(controller).
1619        let should_pull = self.should_call_pull();
1620        // If shouldPull is false, return.
1621        if !should_pull {
1622            return;
1623        }
1624
1625        // If controller.[[pulling]] is true,
1626        if self.pulling.get() {
1627            // Set controller.[[pullAgain]] to true.
1628            self.pull_again.set(true);
1629
1630            // Return.
1631            return;
1632        }
1633
1634        // Assert: controller.[[pullAgain]] is false.
1635        assert!(!self.pull_again.get());
1636
1637        // Set controller.[[pulling]] to true.
1638        self.pulling.set(true);
1639
1640        // Let pullPromise be the result of performing controller.[[pullAlgorithm]].
1641        // Continues into the resolve and reject handling of the native handler.
1642        let global = self.global();
1643        let rooted_controller = DomRoot::from_ref(self);
1644        let controller = Controller::ReadableByteStreamController(rooted_controller.clone());
1645
1646        if let Some(underlying_source) = self.underlying_source.get() {
1647            let handler = PromiseNativeHandler::new(
1648                cx,
1649                &global,
1650                Some(Box::new(PullAlgorithmFulfillmentHandler {
1651                    controller: Dom::from_ref(&rooted_controller),
1652                })),
1653                Some(Box::new(PullAlgorithmRejectionHandler {
1654                    controller: Dom::from_ref(&rooted_controller),
1655                })),
1656            );
1657
1658            let mut realm = enter_auto_realm(cx, &*global);
1659            let cx = &mut realm.current_realm();
1660
1661            let result = underlying_source
1662                .call_pull_algorithm(cx, controller)
1663                .unwrap_or_else(|| {
1664                    let promise = Promise::new_resolved_rooted(cx, &global, ());
1665                    Ok(promise)
1666                });
1667            let promise = result.unwrap_or_else(|error| {
1668                rooted!(&in(cx) let mut rval = UndefinedValue());
1669                // TODO: check if `self.global()` is the right globalscope.
1670                error.to_jsval(cx, &global, rval.handle_mut());
1671                Promise::new_rejected_rooted(cx, &global, rval.handle())
1672            });
1673            promise.append_native_handler(cx, &handler);
1674        }
1675    }
1676
1677    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-should-call-pull>
1678    fn should_call_pull(&self) -> bool {
1679        // Let stream be controller.[[stream]].
1680        // Note: the spec does not assert that stream is not undefined here,
1681        // so we return false if it is.
1682        let stream = self.stream.get().unwrap();
1683
1684        // If stream.[[state]] is not "readable", return false.
1685        if !stream.is_readable() {
1686            return false;
1687        }
1688
1689        // If controller.[[closeRequested]] is true, return false.
1690        if self.close_requested.get() {
1691            return false;
1692        }
1693
1694        // If controller.[[started]] is false, return false.
1695        if !self.started.get() {
1696            return false;
1697        }
1698
1699        // If ! ReadableStreamHasDefaultReader(stream) is true and ! ReadableStreamGetNumReadRequests(stream) > 0
1700        // , return true.
1701        if stream.has_default_reader() && stream.get_num_read_requests() > 0 {
1702            return true;
1703        }
1704
1705        // If ! ReadableStreamHasBYOBReader(stream) is true and ! ReadableStreamGetNumReadIntoRequests(stream) > 0
1706        // , return true.
1707        if stream.has_byob_reader() && stream.get_num_read_into_requests() > 0 {
1708            return true;
1709        }
1710
1711        // Let desiredSize be ! ReadableByteStreamControllerGetDesiredSize(controller).
1712        let desired_size = self.get_desired_size();
1713
1714        // Assert: desiredSize is not null.
1715        assert!(desired_size.is_some());
1716
1717        // If desiredSize > 0, return true.
1718        if desired_size.unwrap() > 0. {
1719            return true;
1720        }
1721
1722        // Return false.
1723        false
1724    }
1725    /// <https://streams.spec.whatwg.org/#set-up-readable-byte-stream-controller>
1726    pub(crate) fn setup(
1727        &self,
1728        cx: &mut JSContext,
1729        global: &GlobalScope,
1730        stream: &ReadableStream,
1731    ) -> Fallible<()> {
1732        // Assert: stream.[[controller]] is undefined.
1733        stream.assert_no_controller();
1734
1735        // If autoAllocateChunkSize is not undefined,
1736        if self.auto_allocate_chunk_size.is_some() {
1737            // Assert: ! IsInteger(autoAllocateChunkSize) is true. Implicit
1738            // Assert: autoAllocateChunkSize is positive. (Implicit by type.)
1739        }
1740
1741        // Set controller.[[stream]] to stream.
1742        self.stream.set(Some(stream));
1743
1744        // Set controller.[[pullAgain]] and controller.[[pulling]] to false.
1745        self.pull_again.set(false);
1746        self.pulling.set(false);
1747
1748        // Set controller.[[byobRequest]] to null.
1749        self.byob_request.set(None);
1750
1751        // Perform ! ResetQueue(controller).
1752        self.reset_queue();
1753
1754        // Set controller.[[closeRequested]] and controller.[[started]] to false.
1755        self.close_requested.set(false);
1756        self.started.set(false);
1757
1758        // Set controller.[[strategyHWM]] to highWaterMark.
1759        // Set controller.[[pullAlgorithm]] to pullAlgorithm.
1760        // Set controller.[[cancelAlgorithm]] to cancelAlgorithm.
1761        // Set controller.[[autoAllocateChunkSize]] to autoAllocateChunkSize.
1762        // Set controller.[[pendingPullIntos]] to a new empty list.
1763        // Note: the above steps are done in `new`.
1764
1765        // Set stream.[[controller]] to controller.
1766        let rooted_byte_controller = DomRoot::from_ref(self);
1767        stream.set_byte_controller(&rooted_byte_controller);
1768
1769        if let Some(underlying_source) = rooted_byte_controller.underlying_source.get() {
1770            // Let startResult be the result of performing startAlgorithm. (This might throw an exception.)
1771            let start_result = underlying_source
1772                .call_start_algorithm(
1773                    cx,
1774                    Controller::ReadableByteStreamController(rooted_byte_controller.clone()),
1775                )
1776                .unwrap_or_else(|| Ok(Promise::new_resolved_rooted(cx, global, ())));
1777
1778            // Let startPromise be a promise resolved with startResult.
1779            let start_promise = start_result?;
1780
1781            // Upon fulfillment of startPromise, Upon rejection of startPromise with reason r,
1782            let handler = PromiseNativeHandler::new(
1783                cx,
1784                global,
1785                Some(Box::new(StartAlgorithmFulfillmentHandler {
1786                    controller: Dom::from_ref(&rooted_byte_controller),
1787                })),
1788                Some(Box::new(StartAlgorithmRejectionHandler {
1789                    controller: Dom::from_ref(&rooted_byte_controller),
1790                })),
1791            );
1792            let mut realm = enter_auto_realm(cx, global);
1793            let cx = &mut realm.current_realm();
1794            start_promise.append_native_handler(cx, &handler);
1795        };
1796
1797        Ok(())
1798    }
1799
1800    // <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontroller-releasesteps
1801    pub(crate) fn perform_release_steps(&self) -> Fallible<()> {
1802        // If this.[[pendingPullIntos]] is not empty,
1803        let mut pending_pull_intos = self.pending_pull_intos.borrow_mut();
1804        if !pending_pull_intos.is_empty() {
1805            // Let firstPendingPullInto be this.[[pendingPullIntos]][0].
1806            let mut first_pending_pull_into = RootedTraceableBox::new(pending_pull_intos.remove(0));
1807
1808            // Set firstPendingPullInto’s reader type to "none".
1809            first_pending_pull_into.reader_type = None;
1810
1811            // Set this.[[pendingPullIntos]] to the list « firstPendingPullInto »
1812            pending_pull_intos.clear();
1813            pending_pull_intos.push(*first_pending_pull_into.into_box());
1814        }
1815        Ok(())
1816    }
1817
1818    /// <https://streams.spec.whatwg.org/#rbs-controller-private-cancel>
1819    pub(crate) fn perform_cancel_steps(
1820        &self,
1821        cx: &mut JSContext,
1822        global: &GlobalScope,
1823        reason: SafeHandleValue,
1824    ) -> RootedPromise {
1825        // Perform ! ReadableByteStreamControllerClearPendingPullIntos(this).
1826        self.clear_pending_pull_intos();
1827
1828        // Perform ! ResetQueue(this).
1829        self.reset_queue();
1830
1831        let underlying_source = self
1832            .underlying_source
1833            .get()
1834            .expect("Controller should have a source when the cancel steps are called into.");
1835
1836        // Let result be the result of performing this.[[cancelAlgorithm]], passing in reason.
1837        let result = underlying_source
1838            .call_cancel_algorithm(cx, global, reason)
1839            .unwrap_or_else(|| {
1840                let promise = Promise::new_rooted(cx, global);
1841                promise.resolve_native(cx, &());
1842                Ok(promise)
1843            });
1844
1845        let promise = result.unwrap_or_else(|error| {
1846            rooted!(&in(cx) let mut rval = UndefinedValue());
1847            error.to_jsval(cx, global, rval.handle_mut());
1848            let promise = Promise::new_rooted(cx, global);
1849            promise.reject_native(cx, &rval.handle());
1850            promise
1851        });
1852
1853        // Perform ! ReadableByteStreamControllerClearAlgorithms(this).
1854        self.clear_algorithms();
1855
1856        // Return result(the promise).
1857        promise
1858    }
1859
1860    /// <https://streams.spec.whatwg.org/#rbs-controller-private-pull>
1861    pub(crate) fn perform_pull_steps(&self, cx: &mut JSContext, read_request: &ReadRequest) {
1862        // Let stream be this.[[stream]].
1863        let stream = self.stream.get().unwrap();
1864
1865        // Assert: ! ReadableStreamHasDefaultReader(stream) is true.
1866        assert!(stream.has_default_reader());
1867
1868        // If this.[[queueTotalSize]] > 0,
1869        if self.queue_total_size.get() > 0.0 {
1870            // Assert: ! ReadableStreamGetNumReadRequests(stream) is 0.
1871            assert_eq!(stream.get_num_read_requests(), 0);
1872
1873            // Perform ! ReadableByteStreamControllerFillReadRequestFromQueue(this, readRequest).
1874            let _ = self.fill_read_request_from_queue(cx, read_request);
1875
1876            // Return.
1877            return;
1878        }
1879
1880        // Let autoAllocateChunkSize be this.[[autoAllocateChunkSize]].
1881        let auto_allocate_chunk_size = self.auto_allocate_chunk_size;
1882
1883        // If autoAllocateChunkSize is not undefined,
1884        if let Some(auto_allocate_chunk_size) = auto_allocate_chunk_size {
1885            // create_array_buffer_with_size
1886            // Let buffer be Construct(%ArrayBuffer%, « autoAllocateChunkSize »).
1887            match create_array_buffer_with_size(cx, auto_allocate_chunk_size as usize) {
1888                Ok(buffer) => {
1889                    // Let pullIntoDescriptor be a new pull-into descriptor with
1890                    // buffer buffer.[[Value]]
1891                    // buffer byte length autoAllocateChunkSize
1892                    // byte offset  0
1893                    // byte length  autoAllocateChunkSize
1894                    // bytes filled  0
1895                    // minimum fill 1
1896                    // element size 1
1897                    // view constructor %Uint8Array%
1898                    // reader type  "default"
1899
1900                    // Append pullIntoDescriptor to this.[[pendingPullIntos]].
1901                    self.pending_pull_intos
1902                        .safe_borrow_mut(cx)
1903                        .push(PullIntoDescriptor {
1904                            buffer: *buffer.into_box(),
1905                            buffer_byte_length: auto_allocate_chunk_size,
1906                            byte_length: auto_allocate_chunk_size,
1907                            byte_offset: 0,
1908                            bytes_filled: Cell::new(0),
1909                            minimum_fill: 1,
1910                            element_size: 1,
1911                            view_constructor: Constructor::Name(Type::Uint8),
1912                            reader_type: Some(ReaderType::Default),
1913                        });
1914                },
1915                Err(error) => {
1916                    // If buffer is an abrupt completion,
1917                    // Perform readRequest’s error steps, given buffer.[[Value]].
1918
1919                    rooted!(&in(cx) let mut rval = UndefinedValue());
1920                    error.to_jsval(cx, &self.global(), rval.handle_mut());
1921                    read_request.error_steps(cx, rval.handle());
1922
1923                    // Return.
1924                    return;
1925                },
1926            }
1927        }
1928
1929        // Perform ! ReadableStreamAddReadRequest(stream, readRequest).
1930        stream.add_read_request(read_request);
1931
1932        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(this).
1933        self.call_pull_if_needed(cx);
1934    }
1935
1936    /// Setting the JS object after the heap has settled down.
1937    pub(crate) fn set_underlying_source_this_object(&self, this_object: HandleObject) {
1938        if let Some(underlying_source) = self.underlying_source.get() {
1939            underlying_source.set_underlying_source_this_object(this_object);
1940        }
1941    }
1942
1943    pub(crate) fn remove_entry(&self) -> QueueEntry {
1944        self.queue
1945            .borrow_mut()
1946            .pop_front()
1947            .expect("Reader must have read request when remove is called into.")
1948    }
1949
1950    pub(crate) fn get_queue_total_size(&self) -> f64 {
1951        self.queue_total_size.get()
1952    }
1953
1954    pub(crate) fn get_pending_pull_intos_size(&self) -> usize {
1955        self.pending_pull_intos.borrow().len()
1956    }
1957}
1958
1959impl ReadableByteStreamControllerMethods<crate::DomTypeHolder> for ReadableByteStreamController {
1960    /// <https://streams.spec.whatwg.org/#rbs-controller-byob-request>
1961    fn GetByobRequest(
1962        &self,
1963        cx: &mut js::context::JSContext,
1964    ) -> Fallible<Option<DomRoot<ReadableStreamBYOBRequest>>> {
1965        // Return ! ReadableByteStreamControllerGetBYOBRequest(this).
1966        self.get_byob_request(cx)
1967    }
1968
1969    /// <https://streams.spec.whatwg.org/#rbs-controller-desired-size>
1970    fn GetDesiredSize(&self) -> Option<f64> {
1971        // Return ! ReadableByteStreamControllerGetDesiredSize(this).
1972        self.get_desired_size()
1973    }
1974
1975    /// <https://streams.spec.whatwg.org/#rbs-controller-close>
1976    fn Close(&self, cx: &mut JSContext) -> Fallible<()> {
1977        // If this.[[closeRequested]] is true, throw a TypeError exception.
1978        if self.close_requested.get() {
1979            return Err(Error::Type(c"closeRequested is true".to_owned()));
1980        }
1981
1982        // If this.[[stream]].[[state]] is not "readable", throw a TypeError exception.
1983        if !self.stream.get().unwrap().is_readable() {
1984            return Err(Error::Type(c"stream is not readable".to_owned()));
1985        }
1986
1987        // Perform ? ReadableByteStreamControllerClose(this).
1988        self.close(cx)
1989    }
1990
1991    /// <https://streams.spec.whatwg.org/#rbs-controller-enqueue>
1992    fn Enqueue(
1993        &self,
1994        cx: &mut JSContext,
1995        chunk: js::gc::CustomAutoRooterGuard<js::typedarray::ArrayBufferView>,
1996    ) -> Fallible<()> {
1997        let chunk = HeapBufferSource::<ArrayBufferViewU8>::from_view(cx, chunk);
1998
1999        // If chunk.[[ByteLength]] is 0, throw a TypeError exception.
2000        if chunk.byte_length() == 0 {
2001            return Err(Error::Type(c"chunk.ByteLength is 0".to_owned()));
2002        }
2003
2004        // If chunk.[[ViewedArrayBuffer]].[[ByteLength]] is 0, throw a TypeError exception.
2005        if chunk.viewed_buffer_array_byte_length(cx) == 0 {
2006            return Err(Error::Type(
2007                c"chunk.ViewedArrayBuffer.ByteLength is 0".to_owned(),
2008            ));
2009        }
2010
2011        // If this.[[closeRequested]] is true, throw a TypeError exception.
2012        if self.close_requested.get() {
2013            return Err(Error::Type(c"closeRequested is true".to_owned()));
2014        }
2015
2016        // If this.[[stream]].[[state]] is not "readable", throw a TypeError exception.
2017        if !self.stream.get().unwrap().is_readable() {
2018            return Err(Error::Type(c"stream is not readable".to_owned()));
2019        }
2020
2021        // Return ? ReadableByteStreamControllerEnqueue(this, chunk).
2022        self.enqueue(cx, chunk)
2023    }
2024
2025    /// <https://streams.spec.whatwg.org/#rbs-controller-error>
2026    fn Error(&self, cx: &mut JSContext, e: SafeHandleValue) -> Fallible<()> {
2027        // Perform ! ReadableByteStreamControllerError(this, e).
2028        self.error(cx, e);
2029        Ok(())
2030    }
2031}