script/dom/stream/

readablebytestreamcontroller.rs

/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

use std::cell::Cell;
use std::cmp::min;
use std::collections::VecDeque;
use std::rc::Rc;

use dom_struct::dom_struct;
use js::context::JSContext;
use js::jsapi::{Heap, Type};
use js::jsval::UndefinedValue;
use js::realm::CurrentRealm;
use js::rust::{HandleObject, HandleValue as SafeHandleValue, HandleValue};
use js::typedarray::{ArrayBufferU8, ArrayBufferViewU8};
use script_bindings::cell::DomRefCell;
use script_bindings::reflector::{Reflector, reflect_dom_object};

use super::readablestreambyobreader::ReadIntoRequest;
use super::readablestreamdefaultreader::ReadRequest;
use super::underlyingsourcecontainer::{UnderlyingSourceContainer, UnderlyingSourceType};
use crate::dom::bindings::buffer_source::{
    Constructor, HeapBufferSource, byte_size, create_array_buffer_with_size,
    create_buffer_source_with_constructor,
};
use crate::dom::bindings::codegen::Bindings::ReadableByteStreamControllerBinding::ReadableByteStreamControllerMethods;
use crate::dom::bindings::codegen::UnionTypes::ReadableStreamDefaultControllerOrReadableByteStreamController as Controller;
use crate::dom::bindings::error::{Error, ErrorToJsval, Fallible};
use crate::dom::bindings::reflector::DomGlobal;
use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
use crate::dom::bindings::trace::RootedTraceableBox;
use crate::dom::globalscope::GlobalScope;
use crate::dom::promise::Promise;
use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
use crate::dom::stream::readablestream::ReadableStream;
use crate::dom::stream::readablestreambyobrequest::ReadableStreamBYOBRequest;
use crate::realms::enter_auto_realm;
use crate::script_runtime::CanGc;

/// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry>
#[derive(JSTraceable, MallocSizeOf)]
#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
pub(crate) struct QueueEntry {
    /// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry-buffer>
    #[ignore_malloc_size_of = "HeapBufferSource"]
    buffer: HeapBufferSource<ArrayBufferU8>,
    /// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry-byte-offset>
    byte_offset: usize,
    /// <https://streams.spec.whatwg.org/#readable-byte-stream-queue-entry-byte-length>
    byte_length: usize,
}

impl js::gc::Rootable for QueueEntry {}

impl QueueEntry {
    pub(crate) fn new(
        buffer: RootedTraceableBox<HeapBufferSource<ArrayBufferU8>>,
        byte_offset: usize,
        byte_length: usize,
    ) -> QueueEntry {
        QueueEntry {
            buffer: *buffer.into_box(),
            byte_offset,
            byte_length,
        }
    }
}

#[derive(Debug, Eq, JSTraceable, MallocSizeOf, PartialEq)]
pub(crate) enum ReaderType {
    /// <https://streams.spec.whatwg.org/#readablestreambyobreader>
    Byob,
    /// <https://streams.spec.whatwg.org/#readablestreamdefaultreader>
    Default,
}

/// <https://streams.spec.whatwg.org/#pull-into-descriptor>
#[derive(Eq, JSTraceable, MallocSizeOf, PartialEq)]
#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
pub(crate) struct PullIntoDescriptor {
    #[ignore_malloc_size_of = "HeapBufferSource"]
    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-buffer>
    buffer: HeapBufferSource<ArrayBufferU8>,
    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-buffer-byte-length>
    buffer_byte_length: u64,
    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-byte-offset>
    byte_offset: u64,
    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-byte-length>
    byte_length: u64,
    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-bytes-filled>
    bytes_filled: Cell<u64>,
    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-minimum-fill>
    minimum_fill: u64,
    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-element-size>
    element_size: u64,
    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-view-constructor>
    view_constructor: Constructor,
    /// <https://streams.spec.whatwg.org/#pull-into-descriptor-reader-type>
    reader_type: Option<ReaderType>,
}

impl js::gc::Rootable for PullIntoDescriptor {}

/// The fulfillment handler for
/// <https://streams.spec.whatwg.org/#dom-underlyingsource-start>
#[derive(Clone, JSTraceable, MallocSizeOf)]
#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
struct StartAlgorithmFulfillmentHandler {
    controller: Dom<ReadableByteStreamController>,
}

impl Callback for StartAlgorithmFulfillmentHandler {
    /// Continuation of <https://streams.spec.whatwg.org/#set-up-readable-byte-stream-controller>
    /// Upon fulfillment of startPromise,
    fn callback(&self, cx: &mut CurrentRealm, _v: HandleValue) {
        // Set controller.[[started]] to true.
        self.controller.started.set(true);

        // Assert: controller.[[pulling]] is false.
        assert!(!self.controller.pulling.get());

        // Assert: controller.[[pullAgain]] is false.
        assert!(!self.controller.pull_again.get());

        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
        self.controller.call_pull_if_needed(cx);
    }
}

/// The rejection handler for
/// <https://streams.spec.whatwg.org/#dom-underlyingsource-start>
#[derive(Clone, JSTraceable, MallocSizeOf)]
#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
struct StartAlgorithmRejectionHandler {
    controller: Dom<ReadableByteStreamController>,
}

impl Callback for StartAlgorithmRejectionHandler {
    /// Continuation of <https://streams.spec.whatwg.org/#set-up-readable-byte-stream-controller>
    /// Upon rejection of startPromise with reason r,
    fn callback(&self, cx: &mut CurrentRealm, v: HandleValue) {
        // Perform ! ReadableByteStreamControllerError(controller, r).
        self.controller.error(cx, v);
    }
}

/// The fulfillment handler for
/// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
#[derive(Clone, JSTraceable, MallocSizeOf)]
#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
struct PullAlgorithmFulfillmentHandler {
    controller: Dom<ReadableByteStreamController>,
}

impl Callback for PullAlgorithmFulfillmentHandler {
    /// Continuation of <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
    /// Upon fulfillment of pullPromise
    fn callback(&self, cx: &mut CurrentRealm, _v: HandleValue) {
        // Set controller.[[pulling]] to false.
        self.controller.pulling.set(false);

        // If controller.[[pullAgain]] is true,
        if self.controller.pull_again.get() {
            // Set controller.[[pullAgain]] to false.
            self.controller.pull_again.set(false);

            // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
            self.controller.call_pull_if_needed(cx);
        }
    }
}

/// The rejection handler for
/// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
#[derive(Clone, JSTraceable, MallocSizeOf)]
#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
struct PullAlgorithmRejectionHandler {
    controller: Dom<ReadableByteStreamController>,
}

impl Callback for PullAlgorithmRejectionHandler {
    /// Continuation of <https://streams.spec.whatwg.org/#readable-stream-byte-controller-call-pull-if-needed>
    /// Upon rejection of pullPromise with reason e.
    fn callback(&self, cx: &mut CurrentRealm, v: HandleValue) {
        // Perform ! ReadableByteStreamControllerError(controller, e).
        self.controller.error(cx, v);
    }
}

/// <https://streams.spec.whatwg.org/#readablebytestreamcontroller>
#[dom_struct]
pub(crate) struct ReadableByteStreamController {
    reflector_: Reflector,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-autoallocatechunksize>
    auto_allocate_chunk_size: Option<u64>,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-stream>
    stream: MutNullableDom<ReadableStream>,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-strategyhwm>
    strategy_hwm: f64,
    /// A mutable reference to the underlying source is used to implement these two
    /// internal slots:
    ///
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pullalgorithm>
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-cancelalgorithm>
    underlying_source: MutNullableDom<UnderlyingSourceContainer>,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-queue>
    queue: DomRefCell<VecDeque<QueueEntry>>,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-queuetotalsize>
    queue_total_size: Cell<f64>,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-byobrequest>
    byob_request: MutNullableDom<ReadableStreamBYOBRequest>,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pendingpullintos>
    pending_pull_intos: DomRefCell<Vec<PullIntoDescriptor>>,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-closerequested>
    close_requested: Cell<bool>,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-started>
    started: Cell<bool>,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pulling>
    pulling: Cell<bool>,
    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller-pullalgorithm>
    pull_again: Cell<bool>,
}

impl ReadableByteStreamController {
    fn new_inherited(
        underlying_source_type: UnderlyingSourceType,
        strategy_hwm: f64,
        global: &GlobalScope,
        can_gc: CanGc,
    ) -> ReadableByteStreamController {
        let underlying_source_container =
            UnderlyingSourceContainer::new(global, underlying_source_type, can_gc);
        let auto_allocate_chunk_size = underlying_source_container.auto_allocate_chunk_size();
        ReadableByteStreamController {
            reflector_: Reflector::new(),
            byob_request: MutNullableDom::new(None),
            stream: MutNullableDom::new(None),
            underlying_source: MutNullableDom::new(Some(&*underlying_source_container)),
            auto_allocate_chunk_size,
            pending_pull_intos: DomRefCell::new(Vec::new()),
            strategy_hwm,
            close_requested: Default::default(),
            queue: DomRefCell::new(Default::default()),
            queue_total_size: Default::default(),
            started: Default::default(),
            pulling: Default::default(),
            pull_again: Default::default(),
        }
    }

    pub(crate) fn new(
        underlying_source_type: UnderlyingSourceType,
        strategy_hwm: f64,
        global: &GlobalScope,
        can_gc: CanGc,
    ) -> DomRoot<ReadableByteStreamController> {
        reflect_dom_object(
            Box::new(ReadableByteStreamController::new_inherited(
                underlying_source_type,
                strategy_hwm,
                global,
                can_gc,
            )),
            global,
            can_gc,
        )
    }

    #[allow(dead_code)]
    pub(crate) fn set_stream(&self, stream: &ReadableStream) {
        self.stream.set(Some(stream))
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-pull-into>
    pub(crate) fn perform_pull_into(
        &self,
        cx: &mut JSContext,
        read_into_request: &ReadIntoRequest,
        view: &HeapBufferSource<ArrayBufferViewU8>,
        min: u64,
    ) {
        // Let stream be controller.[[stream]].
        let stream = self.stream.get().unwrap();

        // Let elementSize be 1.
        let mut element_size = 1;

        // Let ctor be %DataView%.
        let mut ctor = Constructor::DataView;

        // If view has a [[TypedArrayName]] internal slot (i.e., it is not a DataView),
        if view.has_typed_array_name() {
            // Set elementSize to the element size specified in the
            // typed array constructors table for view.[[TypedArrayName]].
            let view_typw = view.get_array_buffer_view_type();
            element_size = byte_size(view_typw);

            // Set ctor to the constructor specified in the typed array constructors table for view.[[TypedArrayName]].
            ctor = Constructor::Name(view_typw);
        }

        // Let minimumFill be min × elementSize.
        let minimum_fill = min * element_size;

        // Assert: minimumFill ≥ 0 and minimumFill ≤ view.[[ByteLength]].
        assert!(minimum_fill <= (view.byte_length() as u64));

        // Assert: the remainder after dividing minimumFill by elementSize is 0.
        assert_eq!(minimum_fill % element_size, 0);

        // Let byteOffset be view.[[ByteOffset]].
        let byte_offset = view.get_byte_offset();

        // Let byteLength be view.[[ByteLength]].
        let byte_length = view.byte_length();

        // Let bufferResult be TransferArrayBuffer(view.[[ViewedArrayBuffer]]).
        match view
            .get_array_buffer_view_buffer(cx.into())
            .transfer_array_buffer(cx.into())
        {
            Ok(buffer) => {
                // Let buffer be bufferResult.[[Value]].
                // Let pullIntoDescriptor be a new pull-into descriptor with
                // buffer   buffer
                // buffer byte length   buffer.[[ArrayBufferByteLength]]
                // byte offset  byteOffset
                // byte length  byteLength
                // bytes filled  0
                // minimum fill minimumFill
                // element size elementSize
                // view constructor ctor
                // reader type  "byob"
                let buffer_byte_length = buffer.byte_length();
                let pull_into_descriptor = RootedTraceableBox::new(PullIntoDescriptor {
                    buffer: *buffer.into_box(),
                    buffer_byte_length: buffer_byte_length as u64,
                    byte_offset: byte_offset as u64,
                    byte_length: byte_length as u64,
                    bytes_filled: Cell::new(0),
                    minimum_fill,
                    element_size,
                    view_constructor: ctor.clone(),
                    reader_type: Some(ReaderType::Byob),
                });

                // If controller.[[pendingPullIntos]] is not empty,
                {
                    let mut pending_pull_intos = self.pending_pull_intos.borrow_mut();
                    if !pending_pull_intos.is_empty() {
                        // Append pullIntoDescriptor to controller.[[pendingPullIntos]].
                        pending_pull_intos.push(*pull_into_descriptor.into_box());

                        // Perform ! ReadableStreamAddReadIntoRequest(stream, readIntoRequest).
                        stream.add_read_into_request(read_into_request);

                        // Return.
                        return;
                    }
                }

                // If stream.[[state]] is "closed",
                if stream.is_closed() {
                    // Let emptyView be ! Construct(ctor, « pullIntoDescriptor’s buffer,
                    // pullIntoDescriptor’s byte offset, 0 »).
                    if let Ok(empty_view) = create_buffer_source_with_constructor(
                        cx,
                        &ctor,
                        &pull_into_descriptor.buffer,
                        pull_into_descriptor.byte_offset as usize,
                        0,
                    ) {
                        // Perform readIntoRequest’s close steps, given emptyView.
                        let result = RootedTraceableBox::new(Heap::default());
                        rooted!(&in(cx) let mut view_value = UndefinedValue());
                        empty_view.get_buffer_view_value(cx.into(), view_value.handle_mut());
                        result.set(*view_value);

                        read_into_request.close_steps(cx, Some(result));

                        // Return.
                        return;
                    } else {
                        return;
                    }
                }

                // If controller.[[queueTotalSize]] > 0,
                if self.queue_total_size.get() > 0.0 {
                    // If ! ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(
                    // controller, pullIntoDescriptor) is true,
                    if self.fill_pull_into_descriptor_from_queue(cx, &pull_into_descriptor) {
                        // Let filledView be ! ReadableByteStreamControllerConvertPullIntoDescriptor(
                        // pullIntoDescriptor).
                        if let Ok(filled_view) =
                            self.convert_pull_into_descriptor(cx, &pull_into_descriptor)
                        {
                            // Perform ! ReadableByteStreamControllerHandleQueueDrain(controller).
                            self.handle_queue_drain(cx);

                            // Perform readIntoRequest’s chunk steps, given filledView.
                            let result = RootedTraceableBox::new(Heap::default());
                            rooted!(&in(cx) let mut view_value = UndefinedValue());
                            filled_view.get_buffer_view_value(cx.into(), view_value.handle_mut());
                            result.set(*view_value);
                            read_into_request.chunk_steps(result, CanGc::from_cx(cx));

                            // Return.
                            return;
                        } else {
                            return;
                        }
                    }

                    // If controller.[[closeRequested]] is true,
                    if self.close_requested.get() {
                        // Let e be a new TypeError exception.
                        rooted!(&in(cx) let mut error = UndefinedValue());
                        Error::Type(c"close requested".to_owned()).to_jsval(
                            cx.into(),
                            &self.global(),
                            error.handle_mut(),
                            CanGc::from_cx(cx),
                        );

                        // Perform ! ReadableByteStreamControllerError(controller, e).
                        self.error(cx, error.handle());

                        // Perform readIntoRequest’s error steps, given e.
                        read_into_request.error_steps(error.handle(), CanGc::from_cx(cx));

                        // Return.
                        return;
                    }
                }

                // Append pullIntoDescriptor to controller.[[pendingPullIntos]].
                {
                    self.pending_pull_intos
                        .borrow_mut()
                        .push(*pull_into_descriptor.into_box());
                }
                // Perform ! ReadableStreamAddReadIntoRequest(stream, readIntoRequest).
                stream.add_read_into_request(read_into_request);

                // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
                self.call_pull_if_needed(cx);
            },
            Err(error) => {
                // If bufferResult is an abrupt completion,

                // Perform readIntoRequest’s error steps, given bufferResult.[[Value]].
                rooted!(&in(cx) let mut rval = UndefinedValue());
                error.to_jsval(
                    cx.into(),
                    &self.global(),
                    rval.handle_mut(),
                    CanGc::from_cx(cx),
                );
                read_into_request.error_steps(rval.handle(), CanGc::from_cx(cx));

                // Return.
            },
        }
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond>
    pub(crate) fn respond(&self, cx: &mut JSContext, bytes_written: u64) -> Fallible<()> {
        {
            // Assert: controller.[[pendingPullIntos]] is not empty.
            let mut pending_pull_intos = self.pending_pull_intos.borrow_mut();
            assert!(!pending_pull_intos.is_empty());

            // Let firstDescriptor be controller.[[pendingPullIntos]][0].
            let first_descriptor = pending_pull_intos.first_mut().unwrap();

            // Let state be controller.[[stream]].[[state]].
            let stream = self.stream.get().unwrap();

            // If state is "closed",
            if stream.is_closed() {
                // If bytesWritten is not 0, throw a TypeError exception.
                if bytes_written != 0 {
                    return Err(Error::Type(
                        c"bytesWritten not zero on closed stream".to_owned(),
                    ));
                }
            } else {
                // Assert: state is "readable".
                assert!(stream.is_readable());

                // If bytesWritten is 0, throw a TypeError exception.
                if bytes_written == 0 {
                    return Err(Error::Type(c"bytesWritten is 0".to_owned()));
                }

                // If firstDescriptor’s bytes filled + bytesWritten > firstDescriptor’s byte length,
                // throw a RangeError exception.
                if first_descriptor.bytes_filled.get() + bytes_written >
                    first_descriptor.byte_length
                {
                    return Err(Error::Range(
                        c"bytes filled + bytesWritten > byte length".to_owned(),
                    ));
                }
            }

            // Set firstDescriptor’s buffer to ! TransferArrayBuffer(firstDescriptor’s buffer).
            first_descriptor.buffer = *first_descriptor
                .buffer
                .transfer_array_buffer(cx.into())
                .expect("TransferArrayBuffer failed")
                .into_box();
        }

        // Perform ? ReadableByteStreamControllerRespondInternal(controller, bytesWritten).
        self.respond_internal(cx, bytes_written)
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-internal>
    fn respond_internal(&self, cx: &mut JSContext, bytes_written: u64) -> Fallible<()> {
        {
            // Let firstDescriptor be controller.[[pendingPullIntos]][0].
            let pending_pull_intos = self.pending_pull_intos.borrow();
            let first_descriptor = pending_pull_intos.first().unwrap();

            // Assert: ! CanTransferArrayBuffer(firstDescriptor’s buffer) is true
            assert!(first_descriptor.buffer.can_transfer_array_buffer(cx.into()));
        }

        // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
        self.invalidate_byob_request();

        // Let state be controller.[[stream]].[[state]].
        let stream = self.stream.get().unwrap();

        // If state is "closed",
        if stream.is_closed() {
            // Assert: bytesWritten is 0.
            assert_eq!(bytes_written, 0);

            // Perform ! ReadableByteStreamControllerRespondInClosedState(controller, firstDescriptor).
            self.respond_in_closed_state(cx)
                .expect("respond_in_closed_state failed");
        } else {
            // Assert: state is "readable".
            assert!(stream.is_readable());

            // Assert: bytesWritten > 0.
            assert!(bytes_written > 0);

            // Perform ? ReadableByteStreamControllerRespondInReadableState(controller, bytesWritten, firstDescriptor).
            self.respond_in_readable_state(cx, bytes_written)?;
        }

        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
        self.call_pull_if_needed(cx);

        Ok(())
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-in-closed-state>
    fn respond_in_closed_state(&self, cx: &mut JSContext) -> Fallible<()> {
        let pending_pull_intos = self.pending_pull_intos.borrow();
        let first_descriptor = pending_pull_intos.first().unwrap();

        // Assert: the remainder after dividing firstDescriptor’s bytes filled
        // by firstDescriptor’s element size is 0.
        assert_eq!(
            first_descriptor.bytes_filled.get() % first_descriptor.element_size,
            0
        );

        // If firstDescriptor’s reader type is "none",
        // perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
        let reader_type = first_descriptor.reader_type.is_none();

        // needed to drop the borrow and avoid BorrowMutError
        drop(pending_pull_intos);

        if reader_type {
            self.shift_pending_pull_into();
        }

        // Let stream be controller.[[stream]].
        let stream = self.stream.get().unwrap();

        // If ! ReadableStreamHasBYOBReader(stream) is true,
        if stream.has_byob_reader() {
            // Let filledPullIntos be a new empty list.
            rooted!(&in(cx) let mut filled_pull_intos = Vec::new());

            // While filledPullIntos’s size < ! ReadableStreamGetNumReadIntoRequests(stream),
            while filled_pull_intos.len() < stream.get_num_read_into_requests() {
                // Let pullIntoDescriptor be ! ReadableByteStreamControllerShiftPendingPullInto(controller).
                // Append pullIntoDescriptor to filledPullIntos.
                filled_pull_intos.push(self.shift_pending_pull_into());
            }

            // For each filledPullInto of filledPullIntos,
            for filled_pull_into in &*filled_pull_intos {
                // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(stream, filledPullInto).
                self.commit_pull_into_descriptor(cx, filled_pull_into)
                    .expect("commit_pull_into_descriptor failed");
            }
        }

        Ok(())
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-in-readable-state>
    fn respond_in_readable_state(&self, cx: &mut JSContext, bytes_written: u64) -> Fallible<()> {
        let pending_pull_intos = self.pending_pull_intos.borrow();
        let first_descriptor = pending_pull_intos.first().unwrap();

        // Assert: pullIntoDescriptor’s bytes filled + bytesWritten ≤ pullIntoDescriptor’s byte length.
        assert!(
            first_descriptor.bytes_filled.get() + bytes_written <= first_descriptor.byte_length
        );

        // Perform ! ReadableByteStreamControllerFillHeadPullIntoDescriptor(
        // controller, bytesWritten, pullIntoDescriptor).
        self.fill_head_pull_into_descriptor(bytes_written, first_descriptor);

        // If pullIntoDescriptor’s reader type is "none",
        if first_descriptor.reader_type.is_none() {
            // needed to drop the borrow and avoid BorrowMutError
            drop(pending_pull_intos);

            // Perform ? ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue(controller, pullIntoDescriptor).
            self.enqueue_detached_pull_into_to_queue(cx)?;

            // Let filledPullIntos be the result of performing
            // ! ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
            rooted!(&in(cx) let filled_pull_intos = self.process_pull_into_descriptors_using_queue(cx));

            // For each filledPullInto of filledPullIntos,
            for filled_pull_into in &*filled_pull_intos {
                // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]]
                // , filledPullInto).
                self.commit_pull_into_descriptor(cx, filled_pull_into)
                    .expect("commit_pull_into_descriptor failed");
            }

            // Return.
            return Ok(());
        }

        // If pullIntoDescriptor’s bytes filled < pullIntoDescriptor’s minimum fill, return.
        if first_descriptor.bytes_filled.get() < first_descriptor.minimum_fill {
            return Ok(());
        }

        // needed to drop the borrow and avoid BorrowMutError
        drop(pending_pull_intos);

        // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
        rooted!(&in(cx) let pull_into_descriptor = self.shift_pending_pull_into());

        // Let remainderSize be the remainder after dividing pullIntoDescriptor’s bytes
        // filled by pullIntoDescriptor’s element size.
        let remainder_size =
            pull_into_descriptor.bytes_filled.get() % pull_into_descriptor.element_size;

        // If remainderSize > 0,
        if remainder_size > 0 {
            // Let end be pullIntoDescriptor’s byte offset + pullIntoDescriptor’s bytes filled.
            let end = pull_into_descriptor.byte_offset + pull_into_descriptor.bytes_filled.get();

            // Perform ? ReadableByteStreamControllerEnqueueClonedChunkToQueue(controller,
            // pullIntoDescriptor’s buffer, end − remainderSize, remainderSize).
            self.enqueue_cloned_chunk_to_queue(
                cx,
                &pull_into_descriptor.buffer,
                end - remainder_size,
                remainder_size,
            )?;
        }

        // Set pullIntoDescriptor’s bytes filled to pullIntoDescriptor’s bytes filled − remainderSize.
        pull_into_descriptor
            .bytes_filled
            .set(pull_into_descriptor.bytes_filled.get() - remainder_size);

        // Let filledPullIntos be the result of performing
        // ! ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
        rooted!(&in(cx) let filled_pull_intos = self.process_pull_into_descriptors_using_queue(cx));

        // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]], pullIntoDescriptor).
        self.commit_pull_into_descriptor(cx, &pull_into_descriptor)
            .expect("commit_pull_into_descriptor failed");

        // For each filledPullInto of filledPullIntos,
        for filled_pull_into in &*filled_pull_intos {
            // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]], filledPullInto).
            self.commit_pull_into_descriptor(cx, filled_pull_into)
                .expect("commit_pull_into_descriptor failed");
        }

        Ok(())
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-respond-with-new-view>
    pub(crate) fn respond_with_new_view(
        &self,
        cx: &mut JSContext,
        view: &HeapBufferSource<ArrayBufferViewU8>,
    ) -> Fallible<()> {
        let view_byte_length;
        {
            // Assert: controller.[[pendingPullIntos]] is not empty.
            let mut pending_pull_intos = self.pending_pull_intos.borrow_mut();
            assert!(!pending_pull_intos.is_empty());

            // Assert: ! IsDetachedBuffer(view.[[ViewedArrayBuffer]]) is false.
            assert!(!view.is_detached_buffer(cx.into()));

            // Let firstDescriptor be controller.[[pendingPullIntos]][0].
            let first_descriptor = pending_pull_intos.first_mut().unwrap();

            // Let state be controller.[[stream]].[[state]].
            let stream = self.stream.get().unwrap();

            // If state is "closed",
            if stream.is_closed() {
                // If view.[[ByteLength]] is not 0, throw a TypeError exception.
                if view.byte_length() != 0 {
                    return Err(Error::Type(c"view byte length is not 0".to_owned()));
                }
            } else {
                // Assert: state is "readable".
                assert!(stream.is_readable());

                // If view.[[ByteLength]] is 0, throw a TypeError exception.
                if view.byte_length() == 0 {
                    return Err(Error::Type(c"view byte length is 0".to_owned()));
                }
            }

            // If firstDescriptor’s byte offset + firstDescriptor’ bytes filled is not view.[[ByteOffset]],
            // throw a RangeError exception.
            if first_descriptor.byte_offset + first_descriptor.bytes_filled.get() !=
                (view.get_byte_offset() as u64)
            {
                return Err(Error::Range(
                    c"firstDescriptor's byte offset + bytes filled is not view byte offset"
                        .to_owned(),
                ));
            }

            // If firstDescriptor’s buffer byte length is not view.[[ViewedArrayBuffer]].[[ByteLength]],
            // throw a RangeError exception.
            if first_descriptor.buffer_byte_length !=
                (view.viewed_buffer_array_byte_length(cx.into()) as u64)
            {
                return Err(Error::Range(
                c"firstDescriptor's buffer byte length is not view viewed buffer array byte length"
                    .to_owned(),
            ));
            }

            // If firstDescriptor’s bytes filled + view.[[ByteLength]] > firstDescriptor’s byte length,
            // throw a RangeError exception.
            if first_descriptor.bytes_filled.get() + (view.byte_length()) as u64 >
                first_descriptor.byte_length
            {
                return Err(Error::Range(
                    c"bytes filled + view byte length > byte length".to_owned(),
                ));
            }

            // Let viewByteLength be view.[[ByteLength]].
            view_byte_length = view.byte_length();

            // Set firstDescriptor’s buffer to ? TransferArrayBuffer(view.[[ViewedArrayBuffer]]).
            first_descriptor.buffer = *view
                .get_array_buffer_view_buffer(cx.into())
                .transfer_array_buffer(cx.into())?
                .into_box();
        }

        // Perform ? ReadableByteStreamControllerRespondInternal(controller, viewByteLength).
        self.respond_internal(cx, view_byte_length as u64)
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-get-desired-size>
    pub(crate) fn get_desired_size(&self) -> Option<f64> {
        // Let state be controller.[[stream]].[[state]].
        let stream = self.stream.get()?;

        // If state is "errored", return null.
        if stream.is_errored() {
            return None;
        }

        // If state is "closed", return 0.
        if stream.is_closed() {
            return Some(0.0);
        }

        // Return controller.[[strategyHWM]] − controller.[[queueTotalSize]].
        Some(self.strategy_hwm - self.queue_total_size.get())
    }

    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollergetbyobrequest>
    pub(crate) fn get_byob_request(
        &self,
        cx: &mut js::context::JSContext,
    ) -> Fallible<Option<DomRoot<ReadableStreamBYOBRequest>>> {
        // If controller.[[byobRequest]] is null and controller.[[pendingPullIntos]] is not empty,
        let pending_pull_intos = self.pending_pull_intos.borrow();
        if self.byob_request.get().is_none() && !pending_pull_intos.is_empty() {
            // Let firstDescriptor be controller.[[pendingPullIntos]][0].
            let first_descriptor = pending_pull_intos.first().unwrap();
            // Let view be ! Construct(%Uint8Array%, « firstDescriptor’s buffer,
            // firstDescriptor’s byte offset + firstDescriptor’s bytes filled,
            // firstDescriptor’s byte length − firstDescriptor’s bytes filled »).

            let byte_offset = first_descriptor.byte_offset + first_descriptor.bytes_filled.get();
            let byte_length = first_descriptor.byte_length - first_descriptor.bytes_filled.get();

            let view = create_buffer_source_with_constructor(
                cx,
                &Constructor::Name(Type::Uint8),
                &first_descriptor.buffer,
                byte_offset as usize,
                byte_length as usize,
            )
            .expect("Construct Uint8Array failed");

            // Let byobRequest be a new ReadableStreamBYOBRequest.
            let byob_request = ReadableStreamBYOBRequest::new(&self.global(), CanGc::from_cx(cx));

            // Set byobRequest.[[controller]] to controller.
            byob_request.set_controller(Some(&DomRoot::from_ref(self)));

            // Set byobRequest.[[view]] to view.
            byob_request.set_view(Some(view));

            // Set controller.[[byobRequest]] to byobRequest.
            self.byob_request.set(Some(&byob_request));
        }

        // Return controller.[[byobRequest]].
        Ok(self.byob_request.get())
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-close>
    pub(crate) fn close(&self, cx: &mut JSContext) -> Fallible<()> {
        // Let stream be controller.[[stream]].
        let stream = self.stream.get().unwrap();

        // If controller.[[closeRequested]] is true or stream.[[state]] is not "readable", return.
        if self.close_requested.get() || !stream.is_readable() {
            return Ok(());
        }

        // If controller.[[queueTotalSize]] > 0,
        if self.queue_total_size.get() > 0.0 {
            // Set controller.[[closeRequested]] to true.
            self.close_requested.set(true);
            // Return.
            return Ok(());
        }

        {
            // If controller.[[pendingPullIntos]] is not empty,
            let pending_pull_intos = self.pending_pull_intos.borrow();
            if !pending_pull_intos.is_empty() {
                // Let firstPendingPullInto be controller.[[pendingPullIntos]][0].
                let first_pending_pull_into = pending_pull_intos.first().unwrap();

                // If the remainder after dividing firstPendingPullInto’s bytes filled by
                // firstPendingPullInto’s element size is not 0,
                if !first_pending_pull_into
                    .bytes_filled
                    .get()
                    .is_multiple_of(first_pending_pull_into.element_size)
                {
                    // needed to drop the borrow and avoid BorrowMutError
                    drop(pending_pull_intos);

                    // Let e be a new TypeError exception.
                    let e = Error::Type(
                        c"remainder after dividing firstPendingPullInto's bytes
                    filled by firstPendingPullInto's element size is not 0"
                            .to_owned(),
                    );

                    // Perform ! ReadableByteStreamControllerError(controller, e).
                    rooted!(&in(cx) let mut error = UndefinedValue());
                    e.clone().to_jsval(
                        cx.into(),
                        &self.global(),
                        error.handle_mut(),
                        CanGc::from_cx(cx),
                    );
                    self.error(cx, error.handle());

                    // Throw e.
                    return Err(e);
                }
            }
        }

        // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
        self.clear_algorithms();

        // Perform ! ReadableStreamClose(stream).
        stream.close(cx);
        Ok(())
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-error>
    pub(crate) fn error(&self, cx: &mut JSContext, e: SafeHandleValue) {
        // Let stream be controller.[[stream]].
        let stream = self.stream.get().unwrap();

        // If stream.[[state]] is not "readable", return.
        if !stream.is_readable() {
            return;
        }

        // Perform ! ReadableByteStreamControllerClearPendingPullIntos(controller).
        self.clear_pending_pull_intos();

        // Perform ! ResetQueue(controller).
        self.reset_queue();

        // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
        self.clear_algorithms();

        // Perform ! ReadableStreamError(stream, e).
        stream.error(cx, e);
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-clear-algorithms>
    fn clear_algorithms(&self) {
        // Set controller.[[pullAlgorithm]] to undefined.
        // Set controller.[[cancelAlgorithm]] to undefined.
        self.underlying_source.set(None);
    }

    /// <https://streams.spec.whatwg.org/#reset-queue>
    pub(crate) fn reset_queue(&self) {
        // Assert: container has [[queue]] and [[queueTotalSize]] internal slots.

        // Set container.[[queue]] to a new empty list.
        self.queue.borrow_mut().clear();

        // Set container.[[queueTotalSize]] to 0.
        self.queue_total_size.set(0.0);
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-clear-pending-pull-intos>
    pub(crate) fn clear_pending_pull_intos(&self) {
        // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
        self.invalidate_byob_request();

        // Set controller.[[pendingPullIntos]] to a new empty list.
        self.pending_pull_intos.borrow_mut().clear();
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-invalidate-byob-request>
    pub(crate) fn invalidate_byob_request(&self) {
        if let Some(byob_request) = self.byob_request.get() {
            // Set controller.[[byobRequest]].[[controller]] to undefined.
            byob_request.set_controller(None);

            // Set controller.[[byobRequest]].[[view]] to null.
            byob_request.set_view(None);

            // Set controller.[[byobRequest]] to null.
            self.byob_request.set(None);
        }
        // If controller.[[byobRequest]] is null, return.
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-enqueue>
    pub(crate) fn enqueue(
        &self,
        cx: &mut JSContext,
        chunk: RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>,
    ) -> Fallible<()> {
        // Let stream be controller.[[stream]].
        let stream = self.stream.get().unwrap();

        // If controller.[[closeRequested]] is true or stream.[[state]] is not "readable", return.
        if self.close_requested.get() || !stream.is_readable() {
            return Ok(());
        }

        // Let buffer be chunk.[[ViewedArrayBuffer]].
        let buffer = chunk.get_array_buffer_view_buffer(cx.into());

        // Let byteOffset be chunk.[[ByteOffset]].
        let byte_offset = chunk.get_byte_offset();

        // Let byteLength be chunk.[[ByteLength]].
        let byte_length = chunk.byte_length();

        // If ! IsDetachedBuffer(buffer) is true, throw a TypeError exception.
        if buffer.is_detached_buffer(cx.into()) {
            return Err(Error::Type(c"buffer is detached".to_owned()));
        }

        // Let transferredBuffer be ? TransferArrayBuffer(buffer).
        let transferred_buffer = buffer.transfer_array_buffer(cx.into())?;

        // If controller.[[pendingPullIntos]] is not empty,
        {
            let mut pending_pull_intos = self.pending_pull_intos.borrow_mut();
            if !pending_pull_intos.is_empty() {
                // Let firstPendingPullInto be controller.[[pendingPullIntos]][0].
                let first_descriptor = pending_pull_intos.first_mut().unwrap();
                // If ! IsDetachedBuffer(firstPendingPullInto’s buffer) is true, throw a TypeError exception.
                if first_descriptor.buffer.is_detached_buffer(cx.into()) {
                    return Err(Error::Type(c"buffer is detached".to_owned()));
                }

                // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
                self.invalidate_byob_request();

                // Set firstPendingPullInto’s buffer to ! TransferArrayBuffer(firstPendingPullInto’s buffer).
                first_descriptor.buffer = *first_descriptor
                    .buffer
                    .transfer_array_buffer(cx.into())
                    .expect("TransferArrayBuffer failed")
                    .into_box();

                // If firstPendingPullInto’s reader type is "none",
                if first_descriptor.reader_type.is_none() {
                    // needed to drop the borrow and avoid BorrowMutError
                    drop(pending_pull_intos);

                    // perform ? ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue(
                    // controller, firstPendingPullInto).
                    self.enqueue_detached_pull_into_to_queue(cx)?;
                }
            }
        }

        // If ! ReadableStreamHasDefaultReader(stream) is true,
        if stream.has_default_reader() {
            // Perform ! ReadableByteStreamControllerProcessReadRequestsUsingQueue(controller).
            self.process_read_requests_using_queue(cx)
                .expect("process_read_requests_using_queue failed");

            // If ! ReadableStreamGetNumReadRequests(stream) is 0,
            if stream.get_num_read_requests() == 0 {
                // Assert: controller.[[pendingPullIntos]] is empty.
                {
                    assert!(self.pending_pull_intos.borrow().is_empty());
                }

                // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue(
                // controller, transferredBuffer, byteOffset, byteLength).
                self.enqueue_chunk_to_queue(transferred_buffer, byte_offset, byte_length);
            } else {
                // Assert: controller.[[queue]] is empty.
                assert!(self.queue.borrow().is_empty());

                // If controller.[[pendingPullIntos]] is not empty,

                let pending_pull_intos = self.pending_pull_intos.borrow();
                if !pending_pull_intos.is_empty() {
                    // Assert: controller.[[pendingPullIntos]][0]'s reader type is "default".
                    assert!(matches!(
                        pending_pull_intos.first().unwrap().reader_type,
                        Some(ReaderType::Default)
                    ));

                    // needed to drop the borrow and avoid BorrowMutError
                    drop(pending_pull_intos);

                    // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
                    self.shift_pending_pull_into();
                }

                // Let transferredView be ! Construct(%Uint8Array%, « transferredBuffer, byteOffset, byteLength »).
                let transferred_view = create_buffer_source_with_constructor(
                    cx,
                    &Constructor::Name(Type::Uint8),
                    &transferred_buffer,
                    byte_offset,
                    byte_length,
                )
                .expect("Construct Uint8Array failed");

                // Perform ! ReadableStreamFulfillReadRequest(stream, transferredView, false).
                rooted!(&in(cx) let mut view_value = UndefinedValue());
                transferred_view.get_buffer_view_value(cx.into(), view_value.handle_mut());
                stream.fulfill_read_request(cx, view_value.handle(), false);
            }
            // Otherwise, if ! ReadableStreamHasBYOBReader(stream) is true,
        } else if stream.has_byob_reader() {
            // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue(
            // controller, transferredBuffer, byteOffset, byteLength).
            self.enqueue_chunk_to_queue(transferred_buffer, byte_offset, byte_length);

            // Let filledPullIntos be the result of performing !
            // ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
            rooted!(&in(cx) let filled_pull_intos = self.process_pull_into_descriptors_using_queue(cx));

            // For each filledPullInto of filledPullIntos,
            // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(stream, filledPullInto).
            for filled_pull_into in &*filled_pull_intos {
                self.commit_pull_into_descriptor(cx, filled_pull_into)
                    .expect("commit_pull_into_descriptor failed");
            }
        } else {
            // Assert: ! IsReadableStreamLocked(stream) is false.
            assert!(!stream.is_locked());

            // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue
            // (controller, transferredBuffer, byteOffset, byteLength).
            self.enqueue_chunk_to_queue(transferred_buffer, byte_offset, byte_length);
        }

        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
        self.call_pull_if_needed(cx);

        Ok(())
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-commit-pull-into-descriptor>
    fn commit_pull_into_descriptor(
        &self,
        cx: &mut JSContext,
        pull_into_descriptor: &PullIntoDescriptor,
    ) -> Fallible<()> {
        // Assert: stream.[[state]] is not "errored".
        let stream = self.stream.get().unwrap();
        assert!(!stream.is_errored());

        // Assert: pullIntoDescriptor.reader type is not "none".
        assert!(pull_into_descriptor.reader_type.is_some());

        // Let done be false.
        let mut done = false;

        // If stream.[[state]] is "closed",
        if stream.is_closed() {
            // Assert: the remainder after dividing pullIntoDescriptor’s bytes filled
            // by pullIntoDescriptor’s element size is 0.
            assert!(
                pull_into_descriptor
                    .bytes_filled
                    .get()
                    .is_multiple_of(pull_into_descriptor.element_size)
            );

            // Set done to true.
            done = true;
        }

        // Let filledView be ! ReadableByteStreamControllerConvertPullIntoDescriptor(pullIntoDescriptor).
        let filled_view = self
            .convert_pull_into_descriptor(cx, pull_into_descriptor)
            .expect("convert_pull_into_descriptor failed");

        rooted!(&in(cx) let mut view_value = UndefinedValue());
        filled_view.get_buffer_view_value(cx.into(), view_value.handle_mut());

        // If pullIntoDescriptor’s reader type is "default",
        if matches!(pull_into_descriptor.reader_type, Some(ReaderType::Default)) {
            // Perform ! ReadableStreamFulfillReadRequest(stream, filledView, done).

            stream.fulfill_read_request(cx, view_value.handle(), done);
        } else {
            // Assert: pullIntoDescriptor’s reader type is "byob".
            assert!(matches!(
                pull_into_descriptor.reader_type,
                Some(ReaderType::Byob)
            ));

            // Perform ! ReadableStreamFulfillReadIntoRequest(stream, filledView, done).
            stream.fulfill_read_into_request(cx, view_value.handle(), done);
        }
        Ok(())
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-convert-pull-into-descriptor>
    pub(crate) fn convert_pull_into_descriptor(
        &self,
        cx: &mut js::context::JSContext,
        pull_into_descriptor: &PullIntoDescriptor,
    ) -> Fallible<RootedTraceableBox<HeapBufferSource<ArrayBufferViewU8>>> {
        // Let bytesFilled be pullIntoDescriptor’s bytes filled.
        let bytes_filled = pull_into_descriptor.bytes_filled.get();

        // Let elementSize be pullIntoDescriptor’s element size.
        let element_size = pull_into_descriptor.element_size;

        // Assert: bytesFilled ≤ pullIntoDescriptor’s byte length.
        assert!(bytes_filled <= pull_into_descriptor.byte_length);

        // Assert: the remainder after dividing bytesFilled by elementSize is 0.
        assert!(bytes_filled.is_multiple_of(element_size));

        // Let buffer be ! TransferArrayBuffer(pullIntoDescriptor’s buffer).
        let buffer = pull_into_descriptor
            .buffer
            .transfer_array_buffer(cx.into())
            .expect("TransferArrayBuffer failed");

        // Return ! Construct(pullIntoDescriptor’s view constructor,
        // « buffer, pullIntoDescriptor’s byte offset, bytesFilled ÷ elementSize »).
        Ok(create_buffer_source_with_constructor(
            cx,
            &pull_into_descriptor.view_constructor,
            &buffer,
            pull_into_descriptor.byte_offset as usize,
            (bytes_filled / element_size) as usize,
        )
        .expect("Construct view failed"))
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-process-pull-into-descriptors-using-queue>
    pub(crate) fn process_pull_into_descriptors_using_queue(
        &self,
        cx: &mut js::context::JSContext,
    ) -> Vec<PullIntoDescriptor> {
        // Assert: controller.[[closeRequested]] is false.
        assert!(!self.close_requested.get());

        // Let filledPullIntos be a new empty list.
        rooted!(&in(cx) let mut filled_pull_intos = Vec::new());

        // While controller.[[pendingPullIntos]] is not empty,
        loop {
            // If controller.[[queueTotalSize]] is 0, then break.
            if self.queue_total_size.get() == 0.0 {
                break;
            }

            // Let pullIntoDescriptor be controller.[[pendingPullIntos]][0].
            let fill_pull_result = {
                let pending_pull_intos = self.pending_pull_intos.borrow();
                let Some(pull_into_descriptor) = pending_pull_intos.first() else {
                    break;
                };
                self.fill_pull_into_descriptor_from_queue(cx, pull_into_descriptor)
            };

            // If ! ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller, pullIntoDescriptor) is true,
            if fill_pull_result {
                // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
                // Append pullIntoDescriptor to filledPullIntos.
                filled_pull_intos.push(self.shift_pending_pull_into());
            }
        }

        // Return filledPullIntos.
        filled_pull_intos.take()
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-fill-pull-into-descriptor-from-queue>
    pub(crate) fn fill_pull_into_descriptor_from_queue(
        &self,
        cx: &mut js::context::JSContext,
        pull_into_descriptor: &PullIntoDescriptor,
    ) -> bool {
        // Let maxBytesToCopy be min(controller.[[queueTotalSize]],
        // pullIntoDescriptor’s byte length − pullIntoDescriptor’s bytes filled).
        let max_bytes_to_copy = min(
            self.queue_total_size.get() as usize,
            (pull_into_descriptor.byte_length - pull_into_descriptor.bytes_filled.get()) as usize,
        );

        // Let maxBytesFilled be pullIntoDescriptor’s bytes filled + maxBytesToCopy.
        let max_bytes_filled = pull_into_descriptor.bytes_filled.get() as usize + max_bytes_to_copy;

        // Let totalBytesToCopyRemaining be maxBytesToCopy.
        let mut total_bytes_to_copy_remaining = max_bytes_to_copy;

        // Let ready be false.
        let mut ready = false;

        // Assert: ! IsDetachedBuffer(pullIntoDescriptor’s buffer) is false.
        assert!(!pull_into_descriptor.buffer.is_detached_buffer(cx.into()));

        // Assert: pullIntoDescriptor’s bytes filled < pullIntoDescriptor’s minimum fill.
        assert!(pull_into_descriptor.bytes_filled.get() < pull_into_descriptor.minimum_fill);

        // Let remainderBytes be the remainder after dividing maxBytesFilled by pullIntoDescriptor’s element size.
        let remainder_bytes = max_bytes_filled % pull_into_descriptor.element_size as usize;

        // Let maxAlignedBytes be maxBytesFilled − remainderBytes.
        let max_aligned_bytes = max_bytes_filled - remainder_bytes;

        // If maxAlignedBytes ≥ pullIntoDescriptor’s minimum fill,
        if max_aligned_bytes >= pull_into_descriptor.minimum_fill as usize {
            // Set totalBytesToCopyRemaining to maxAlignedBytes − pullIntoDescriptor’s bytes filled.
            total_bytes_to_copy_remaining =
                max_aligned_bytes - (pull_into_descriptor.bytes_filled.get() as usize);

            // Set ready to true.
            ready = true;
        }

        // Let queue be controller.[[queue]].
        let mut queue = self.queue.borrow_mut();

        // While totalBytesToCopyRemaining > 0,
        while total_bytes_to_copy_remaining > 0 {
            // Let headOfQueue be queue[0].
            let head_of_queue = queue.front_mut().unwrap();

            // Let bytesToCopy be min(totalBytesToCopyRemaining, headOfQueue’s byte length).
            let bytes_to_copy = total_bytes_to_copy_remaining.min(head_of_queue.byte_length);

            // Let destStart be pullIntoDescriptor’s byte offset + pullIntoDescriptor’s bytes filled.
            let dest_start =
                pull_into_descriptor.byte_offset + pull_into_descriptor.bytes_filled.get();

            // Let descriptorBuffer be pullIntoDescriptor’s buffer.
            let descriptor_buffer = &pull_into_descriptor.buffer;

            // Let queueBuffer be headOfQueue’s buffer.
            let queue_buffer = &head_of_queue.buffer;

            // Let queueByteOffset be headOfQueue’s byte offset.
            let queue_byte_offset = head_of_queue.byte_offset;

            // Assert: ! CanCopyDataBlockBytes(descriptorBuffer, destStart,
            // queueBuffer, queueByteOffset, bytesToCopy) is true.
            assert!(descriptor_buffer.can_copy_data_block_bytes(
                cx.into(),
                dest_start as usize,
                queue_buffer,
                queue_byte_offset,
                bytes_to_copy
            ));

            // Perform ! CopyDataBlockBytes(descriptorBuffer.[[ArrayBufferData]], destStart,
            // queueBuffer.[[ArrayBufferData]], queueByteOffset, bytesToCopy).
            descriptor_buffer.copy_data_block_bytes(
                cx.into(),
                dest_start as usize,
                queue_buffer,
                queue_byte_offset,
                bytes_to_copy,
            );

            // If headOfQueue’s byte length is bytesToCopy,
            if head_of_queue.byte_length == bytes_to_copy {
                // Remove queue[0].
                queue.pop_front().unwrap();
            } else {
                // Set headOfQueue’s byte offset to headOfQueue’s byte offset + bytesToCopy.
                head_of_queue.byte_offset += bytes_to_copy;

                // Set headOfQueue’s byte length to headOfQueue’s byte length − bytesToCopy.
                head_of_queue.byte_length -= bytes_to_copy;
            }

            // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] − bytesToCopy.
            self.queue_total_size
                .set(self.queue_total_size.get() - (bytes_to_copy as f64));

            // Perform ! ReadableByteStreamControllerFillHeadPullIntoDescriptor(
            // controller, bytesToCopy, pullIntoDescriptor).
            self.fill_head_pull_into_descriptor(bytes_to_copy as u64, pull_into_descriptor);

            // Set totalBytesToCopyRemaining to totalBytesToCopyRemaining − bytesToCopy.
            total_bytes_to_copy_remaining -= bytes_to_copy;
        }

        // If ready is false,
        if !ready {
            // Assert: controller.[[queueTotalSize]] is 0.
            assert!(self.queue_total_size.get() == 0.0);

            // Assert: pullIntoDescriptor’s bytes filled > 0.
            assert!(pull_into_descriptor.bytes_filled.get() > 0);

            // Assert: pullIntoDescriptor’s bytes filled < pullIntoDescriptor’s minimum fill.
            assert!(pull_into_descriptor.bytes_filled.get() < pull_into_descriptor.minimum_fill);
        }

        // Return ready.
        ready
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-fill-head-pull-into-descriptor>
    pub(crate) fn fill_head_pull_into_descriptor(
        &self,
        bytes_copied: u64,
        pull_into_descriptor: &PullIntoDescriptor,
    ) {
        // Assert: either controller.[[pendingPullIntos]] is empty,
        // or controller.[[pendingPullIntos]][0] is pullIntoDescriptor.
        {
            let pending_pull_intos = self.pending_pull_intos.borrow();
            assert!(
                pending_pull_intos.is_empty() ||
                    pending_pull_intos.first().unwrap() == pull_into_descriptor
            );
        }

        // Assert: controller.[[byobRequest]] is null.
        assert!(self.byob_request.get().is_none());

        // Set pullIntoDescriptor’s bytes filled to bytes filled + size.
        pull_into_descriptor
            .bytes_filled
            .set(pull_into_descriptor.bytes_filled.get() + bytes_copied);
    }

    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerenqueuedetachedpullintotoqueue>
    pub(crate) fn enqueue_detached_pull_into_to_queue(&self, cx: &mut JSContext) -> Fallible<()> {
        // first_descriptor: &PullIntoDescriptor,
        let pending_pull_intos = self.pending_pull_intos.borrow();
        let first_descriptor = pending_pull_intos.first().unwrap();

        // Assert: pullIntoDescriptor’s reader type is "none".
        assert!(first_descriptor.reader_type.is_none());

        // If pullIntoDescriptor’s bytes filled > 0, perform ?
        // ReadableByteStreamControllerEnqueueClonedChunkToQueue(controller,
        // pullIntoDescriptor’s buffer, pullIntoDescriptor’s byte offset, pullIntoDescriptor’s bytes filled).

        if first_descriptor.bytes_filled.get() > 0 {
            self.enqueue_cloned_chunk_to_queue(
                cx,
                &first_descriptor.buffer,
                first_descriptor.byte_offset,
                first_descriptor.bytes_filled.get(),
            )?;
        }

        // needed to drop the borrow and avoid BorrowMutError
        drop(pending_pull_intos);

        // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
        self.shift_pending_pull_into();

        Ok(())
    }

    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerenqueueclonedchunktoqueue>
    pub(crate) fn enqueue_cloned_chunk_to_queue(
        &self,
        cx: &mut JSContext,
        buffer: &HeapBufferSource<ArrayBufferU8>,
        byte_offset: u64,
        byte_length: u64,
    ) -> Fallible<()> {
        // Let cloneResult be CloneArrayBuffer(buffer, byteOffset, byteLength, %ArrayBuffer%).
        if let Ok(clone_result) =
            buffer.clone_array_buffer(cx, byte_offset as usize, byte_length as usize)
        {
            // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue
            // (controller, cloneResult.[[Value]], 0, byteLength).
            self.enqueue_chunk_to_queue(clone_result, 0, byte_length as usize);

            Ok(())
        } else {
            // If cloneResult is an abrupt completion,

            // Perform ! ReadableByteStreamControllerError(controller, cloneResult.[[Value]]).
            rooted!(&in(cx) let mut rval = UndefinedValue());
            let error = Error::Type(c"can not clone array buffer".to_owned());
            error.clone().to_jsval(
                cx.into(),
                &self.global(),
                rval.handle_mut(),
                CanGc::from_cx(cx),
            );
            self.error(cx, rval.handle());

            // Return cloneResult.
            Err(error)
        }
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-enqueue-chunk-to-queue>
    pub(crate) fn enqueue_chunk_to_queue(
        &self,
        buffer: RootedTraceableBox<HeapBufferSource<ArrayBufferU8>>,
        byte_offset: usize,
        byte_length: usize,
    ) {
        // Let entry be a new ReadableByteStreamQueueEntry object.
        // Append entry to controller.[[queue]].
        self.queue
            .borrow_mut()
            .push_back(QueueEntry::new(buffer, byte_offset, byte_length));

        // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] + byteLength.
        self.queue_total_size
            .set(self.queue_total_size.get() + byte_length as f64);
    }

    pub(crate) fn in_memory(&self) -> bool {
        let Some(underlying_source) = self.underlying_source.get() else {
            return false;
        };
        underlying_source.in_memory()
    }

    pub(crate) fn get_in_memory_bytes(&self) -> Option<Vec<u8>> {
        let underlying_source = self.underlying_source.get()?;
        if !underlying_source.in_memory() {
            return None;
        }

        let cx = GlobalScope::get_cx();
        self.queue.borrow().iter().try_fold(
            Vec::with_capacity(self.queue_total_size.get() as usize),
            |mut bytes, entry| {
                let mut chunk = vec![0; entry.byte_length];
                entry
                    .buffer
                    .copy_data_to(
                        cx,
                        &mut chunk,
                        entry.byte_offset,
                        entry.byte_offset + entry.byte_length,
                    )
                    .ok()?;
                bytes.extend(chunk);
                Some(bytes)
            },
        )
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-shift-pending-pull-into>
    pub(crate) fn shift_pending_pull_into(&self) -> PullIntoDescriptor {
        // Assert: controller.[[byobRequest]] is null.
        assert!(self.byob_request.get().is_none());

        // Let descriptor be controller.[[pendingPullIntos]][0].
        // Remove descriptor from controller.[[pendingPullIntos]].
        // Return descriptor.
        self.pending_pull_intos.borrow_mut().remove(0)
    }

    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerprocessreadrequestsusingqueue>
    pub(crate) fn process_read_requests_using_queue(&self, cx: &mut JSContext) -> Fallible<()> {
        // Let reader be controller.[[stream]].[[reader]].
        // Assert: reader implements ReadableStreamDefaultReader.
        let reader = self.stream.get().unwrap().get_default_reader();

        // Step 3
        reader.process_read_requests(cx, DomRoot::from_ref(self))
    }

    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerfillreadrequestfromqueue>
    pub(crate) fn fill_read_request_from_queue(
        &self,
        cx: &mut JSContext,
        read_request: &ReadRequest,
    ) -> Fallible<()> {
        // Assert: controller.[[queueTotalSize]] > 0.
        assert!(self.queue_total_size.get() > 0.0);
        // Also assert that the queue has a non-zero length;
        assert!(!self.queue.borrow().is_empty());

        // Let entry be controller.[[queue]][0].
        // Remove entry from controller.[[queue]].
        rooted!(&in(cx) let entry = self.remove_entry());

        // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] − entry’s byte length.
        self.queue_total_size
            .set(self.queue_total_size.get() - entry.byte_length as f64);

        // Perform ! ReadableByteStreamControllerHandleQueueDrain(controller).
        self.handle_queue_drain(cx);

        // Let view be ! Construct(%Uint8Array%, « entry’s buffer, entry’s byte offset, entry’s byte length »).
        let view = create_buffer_source_with_constructor(
            cx,
            &Constructor::Name(Type::Uint8),
            &entry.buffer,
            entry.byte_offset,
            entry.byte_length,
        )
        .expect("Construct Uint8Array failed");

        // Perform readRequest’s chunk steps, given view.
        let result = RootedTraceableBox::new(Heap::default());
        rooted!(&in(cx) let mut view_value = UndefinedValue());
        view.get_buffer_view_value(cx.into(), view_value.handle_mut());
        result.set(*view_value);

        read_request.chunk_steps(cx, result, &self.global());

        Ok(())
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-handle-queue-drain>
    pub(crate) fn handle_queue_drain(&self, cx: &mut JSContext) {
        // Assert: controller.[[stream]].[[state]] is "readable".
        assert!(self.stream.get().unwrap().is_readable());

        // If controller.[[queueTotalSize]] is 0 and controller.[[closeRequested]] is true,
        if self.queue_total_size.get() == 0.0 && self.close_requested.get() {
            // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
            self.clear_algorithms();

            // Perform ! ReadableStreamClose(controller.[[stream]]).
            self.stream.get().unwrap().close(cx);
        } else {
            // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
            self.call_pull_if_needed(cx);
        }
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-call-pull-if-needed>
    fn call_pull_if_needed(&self, cx: &mut JSContext) {
        // Let shouldPull be ! ReadableByteStreamControllerShouldCallPull(controller).
        let should_pull = self.should_call_pull();
        // If shouldPull is false, return.
        if !should_pull {
            return;
        }

        // If controller.[[pulling]] is true,
        if self.pulling.get() {
            // Set controller.[[pullAgain]] to true.
            self.pull_again.set(true);

            // Return.
            return;
        }

        // Assert: controller.[[pullAgain]] is false.
        assert!(!self.pull_again.get());

        // Set controller.[[pulling]] to true.
        self.pulling.set(true);

        // Let pullPromise be the result of performing controller.[[pullAlgorithm]].
        // Continues into the resolve and reject handling of the native handler.
        let global = self.global();
        let rooted_controller = DomRoot::from_ref(self);
        let controller = Controller::ReadableByteStreamController(rooted_controller.clone());

        if let Some(underlying_source) = self.underlying_source.get() {
            let handler = PromiseNativeHandler::new(
                cx,
                &global,
                Some(Box::new(PullAlgorithmFulfillmentHandler {
                    controller: Dom::from_ref(&rooted_controller),
                })),
                Some(Box::new(PullAlgorithmRejectionHandler {
                    controller: Dom::from_ref(&rooted_controller),
                })),
            );

            let mut realm = enter_auto_realm(cx, &*global);
            let cx = &mut realm.current_realm();

            let result = underlying_source
                .call_pull_algorithm(cx, controller)
                .unwrap_or_else(|| {
                    let promise = Promise::new_resolved(&global, cx.into(), (), CanGc::from_cx(cx));
                    Ok(promise)
                });
            let promise = result.unwrap_or_else(|error| {
                rooted!(&in(cx) let mut rval = UndefinedValue());
                // TODO: check if `self.global()` is the right globalscope.
                error.to_jsval(cx.into(), &global, rval.handle_mut(), CanGc::from_cx(cx));
                Promise::new_rejected(&global, cx.into(), rval.handle(), CanGc::from_cx(cx))
            });
            promise.append_native_handler(cx, &handler);
        }
    }

    /// <https://streams.spec.whatwg.org/#readable-byte-stream-controller-should-call-pull>
    fn should_call_pull(&self) -> bool {
        // Let stream be controller.[[stream]].
        // Note: the spec does not assert that stream is not undefined here,
        // so we return false if it is.
        let stream = self.stream.get().unwrap();

        // If stream.[[state]] is not "readable", return false.
        if !stream.is_readable() {
            return false;
        }

        // If controller.[[closeRequested]] is true, return false.
        if self.close_requested.get() {
            return false;
        }

        // If controller.[[started]] is false, return false.
        if !self.started.get() {
            return false;
        }

        // If ! ReadableStreamHasDefaultReader(stream) is true and ! ReadableStreamGetNumReadRequests(stream) > 0
        // , return true.
        if stream.has_default_reader() && stream.get_num_read_requests() > 0 {
            return true;
        }

        // If ! ReadableStreamHasBYOBReader(stream) is true and ! ReadableStreamGetNumReadIntoRequests(stream) > 0
        // , return true.
        if stream.has_byob_reader() && stream.get_num_read_into_requests() > 0 {
            return true;
        }

        // Let desiredSize be ! ReadableByteStreamControllerGetDesiredSize(controller).
        let desired_size = self.get_desired_size();

        // Assert: desiredSize is not null.
        assert!(desired_size.is_some());

        // If desiredSize > 0, return true.
        if desired_size.unwrap() > 0. {
            return true;
        }

        // Return false.
        false
    }
    /// <https://streams.spec.whatwg.org/#set-up-readable-byte-stream-controller>
    pub(crate) fn setup(
        &self,
        cx: &mut JSContext,
        global: &GlobalScope,
        stream: DomRoot<ReadableStream>,
    ) -> Fallible<()> {
        // Assert: stream.[[controller]] is undefined.
        stream.assert_no_controller();

        // If autoAllocateChunkSize is not undefined,
        if self.auto_allocate_chunk_size.is_some() {
            // Assert: ! IsInteger(autoAllocateChunkSize) is true. Implicit
            // Assert: autoAllocateChunkSize is positive. (Implicit by type.)
        }

        // Set controller.[[stream]] to stream.
        self.stream.set(Some(&stream));

        // Set controller.[[pullAgain]] and controller.[[pulling]] to false.
        self.pull_again.set(false);
        self.pulling.set(false);

        // Set controller.[[byobRequest]] to null.
        self.byob_request.set(None);

        // Perform ! ResetQueue(controller).
        self.reset_queue();

        // Set controller.[[closeRequested]] and controller.[[started]] to false.
        self.close_requested.set(false);
        self.started.set(false);

        // Set controller.[[strategyHWM]] to highWaterMark.
        // Set controller.[[pullAlgorithm]] to pullAlgorithm.
        // Set controller.[[cancelAlgorithm]] to cancelAlgorithm.
        // Set controller.[[autoAllocateChunkSize]] to autoAllocateChunkSize.
        // Set controller.[[pendingPullIntos]] to a new empty list.
        // Note: the above steps are done in `new`.

        // Set stream.[[controller]] to controller.
        let rooted_byte_controller = DomRoot::from_ref(self);
        stream.set_byte_controller(&rooted_byte_controller);

        if let Some(underlying_source) = rooted_byte_controller.underlying_source.get() {
            // Let startResult be the result of performing startAlgorithm. (This might throw an exception.)
            let start_result = underlying_source
                .call_start_algorithm(
                    cx,
                    Controller::ReadableByteStreamController(rooted_byte_controller.clone()),
                )
                .unwrap_or_else(|| {
                    Ok(Promise::new_resolved(
                        global,
                        cx.into(),
                        (),
                        CanGc::from_cx(cx),
                    ))
                });

            // Let startPromise be a promise resolved with startResult.
            let start_promise = start_result?;

            // Upon fulfillment of startPromise, Upon rejection of startPromise with reason r,
            let handler = PromiseNativeHandler::new(
                cx,
                global,
                Some(Box::new(StartAlgorithmFulfillmentHandler {
                    controller: Dom::from_ref(&rooted_byte_controller),
                })),
                Some(Box::new(StartAlgorithmRejectionHandler {
                    controller: Dom::from_ref(&rooted_byte_controller),
                })),
            );
            let mut realm = enter_auto_realm(cx, global);
            let cx = &mut realm.current_realm();
            start_promise.append_native_handler(cx, &handler);
        };

        Ok(())
    }

    // <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontroller-releasesteps
    pub(crate) fn perform_release_steps(&self) -> Fallible<()> {
        // If this.[[pendingPullIntos]] is not empty,
        let mut pending_pull_intos = self.pending_pull_intos.borrow_mut();
        if !pending_pull_intos.is_empty() {
            // Let firstPendingPullInto be this.[[pendingPullIntos]][0].
            let mut first_pending_pull_into = RootedTraceableBox::new(pending_pull_intos.remove(0));

            // Set firstPendingPullInto’s reader type to "none".
            first_pending_pull_into.reader_type = None;

            // Set this.[[pendingPullIntos]] to the list « firstPendingPullInto »
            pending_pull_intos.clear();
            pending_pull_intos.push(*first_pending_pull_into.into_box());
        }
        Ok(())
    }

    /// <https://streams.spec.whatwg.org/#rbs-controller-private-cancel>
    pub(crate) fn perform_cancel_steps(
        &self,
        cx: &mut JSContext,
        global: &GlobalScope,
        reason: SafeHandleValue,
    ) -> Rc<Promise> {
        // Perform ! ReadableByteStreamControllerClearPendingPullIntos(this).
        self.clear_pending_pull_intos();

        // Perform ! ResetQueue(this).
        self.reset_queue();

        let underlying_source = self
            .underlying_source
            .get()
            .expect("Controller should have a source when the cancel steps are called into.");

        // Let result be the result of performing this.[[cancelAlgorithm]], passing in reason.
        let result = underlying_source
            .call_cancel_algorithm(cx, global, reason)
            .unwrap_or_else(|| {
                let promise = Promise::new2(cx, global);
                promise.resolve_native_with_cx(cx, &());
                Ok(promise)
            });

        let promise = result.unwrap_or_else(|error| {
            rooted!(&in(cx) let mut rval = UndefinedValue());
            error.to_jsval(cx.into(), global, rval.handle_mut(), CanGc::from_cx(cx));
            let promise = Promise::new2(cx, global);
            promise.reject_native_with_cx(cx, &rval.handle());
            promise
        });

        // Perform ! ReadableByteStreamControllerClearAlgorithms(this).
        self.clear_algorithms();

        // Return result(the promise).
        promise
    }

    /// <https://streams.spec.whatwg.org/#rbs-controller-private-pull>
    pub(crate) fn perform_pull_steps(&self, cx: &mut JSContext, read_request: &ReadRequest) {
        // Let stream be this.[[stream]].
        let stream = self.stream.get().unwrap();

        // Assert: ! ReadableStreamHasDefaultReader(stream) is true.
        assert!(stream.has_default_reader());

        // If this.[[queueTotalSize]] > 0,
        if self.queue_total_size.get() > 0.0 {
            // Assert: ! ReadableStreamGetNumReadRequests(stream) is 0.
            assert_eq!(stream.get_num_read_requests(), 0);

            // Perform ! ReadableByteStreamControllerFillReadRequestFromQueue(this, readRequest).
            let _ = self.fill_read_request_from_queue(cx, read_request);

            // Return.
            return;
        }

        // Let autoAllocateChunkSize be this.[[autoAllocateChunkSize]].
        let auto_allocate_chunk_size = self.auto_allocate_chunk_size;

        // If autoAllocateChunkSize is not undefined,
        if let Some(auto_allocate_chunk_size) = auto_allocate_chunk_size {
            // create_array_buffer_with_size
            // Let buffer be Construct(%ArrayBuffer%, « autoAllocateChunkSize »).
            match create_array_buffer_with_size(cx, auto_allocate_chunk_size as usize) {
                Ok(buffer) => {
                    // Let pullIntoDescriptor be a new pull-into descriptor with
                    // buffer buffer.[[Value]]
                    // buffer byte length autoAllocateChunkSize
                    // byte offset  0
                    // byte length  autoAllocateChunkSize
                    // bytes filled  0
                    // minimum fill 1
                    // element size 1
                    // view constructor %Uint8Array%
                    // reader type  "default"

                    // Append pullIntoDescriptor to this.[[pendingPullIntos]].
                    self.pending_pull_intos
                        .borrow_mut()
                        .push(PullIntoDescriptor {
                            buffer: *buffer.into_box(),
                            buffer_byte_length: auto_allocate_chunk_size,
                            byte_length: auto_allocate_chunk_size,
                            byte_offset: 0,
                            bytes_filled: Cell::new(0),
                            minimum_fill: 1,
                            element_size: 1,
                            view_constructor: Constructor::Name(Type::Uint8),
                            reader_type: Some(ReaderType::Default),
                        });
                },
                Err(error) => {
                    // If buffer is an abrupt completion,
                    // Perform readRequest’s error steps, given buffer.[[Value]].

                    rooted!(&in(cx) let mut rval = UndefinedValue());
                    error.to_jsval(
                        cx.into(),
                        &self.global(),
                        rval.handle_mut(),
                        CanGc::from_cx(cx),
                    );
                    read_request.error_steps(cx, rval.handle());

                    // Return.
                    return;
                },
            }
        }

        // Perform ! ReadableStreamAddReadRequest(stream, readRequest).
        stream.add_read_request(read_request);

        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(this).
        self.call_pull_if_needed(cx);
    }

    /// Setting the JS object after the heap has settled down.
    pub(crate) fn set_underlying_source_this_object(&self, this_object: HandleObject) {
        if let Some(underlying_source) = self.underlying_source.get() {
            underlying_source.set_underlying_source_this_object(this_object);
        }
    }

    pub(crate) fn remove_entry(&self) -> QueueEntry {
        self.queue
            .borrow_mut()
            .pop_front()
            .expect("Reader must have read request when remove is called into.")
    }

    pub(crate) fn get_queue_total_size(&self) -> f64 {
        self.queue_total_size.get()
    }

    pub(crate) fn get_pending_pull_intos_size(&self) -> usize {
        self.pending_pull_intos.borrow().len()
    }
}

impl ReadableByteStreamControllerMethods<crate::DomTypeHolder> for ReadableByteStreamController {
    /// <https://streams.spec.whatwg.org/#rbs-controller-byob-request>
    fn GetByobRequest(
        &self,
        cx: &mut js::context::JSContext,
    ) -> Fallible<Option<DomRoot<ReadableStreamBYOBRequest>>> {
        // Return ! ReadableByteStreamControllerGetBYOBRequest(this).
        self.get_byob_request(cx)
    }

    /// <https://streams.spec.whatwg.org/#rbs-controller-desired-size>
    fn GetDesiredSize(&self) -> Option<f64> {
        // Return ! ReadableByteStreamControllerGetDesiredSize(this).
        self.get_desired_size()
    }

    /// <https://streams.spec.whatwg.org/#rbs-controller-close>
    fn Close(&self, cx: &mut JSContext) -> Fallible<()> {
        // If this.[[closeRequested]] is true, throw a TypeError exception.
        if self.close_requested.get() {
            return Err(Error::Type(c"closeRequested is true".to_owned()));
        }

        // If this.[[stream]].[[state]] is not "readable", throw a TypeError exception.
        if !self.stream.get().unwrap().is_readable() {
            return Err(Error::Type(c"stream is not readable".to_owned()));
        }

        // Perform ? ReadableByteStreamControllerClose(this).
        self.close(cx)
    }

    /// <https://streams.spec.whatwg.org/#rbs-controller-enqueue>
    fn Enqueue(
        &self,
        cx: &mut JSContext,
        chunk: js::gc::CustomAutoRooterGuard<js::typedarray::ArrayBufferView>,
    ) -> Fallible<()> {
        let chunk = HeapBufferSource::<ArrayBufferViewU8>::from_view(chunk);

        // If chunk.[[ByteLength]] is 0, throw a TypeError exception.
        if chunk.byte_length() == 0 {
            return Err(Error::Type(c"chunk.ByteLength is 0".to_owned()));
        }

        // If chunk.[[ViewedArrayBuffer]].[[ByteLength]] is 0, throw a TypeError exception.
        if chunk.viewed_buffer_array_byte_length(cx.into()) == 0 {
            return Err(Error::Type(
                c"chunk.ViewedArrayBuffer.ByteLength is 0".to_owned(),
            ));
        }

        // If this.[[closeRequested]] is true, throw a TypeError exception.
        if self.close_requested.get() {
            return Err(Error::Type(c"closeRequested is true".to_owned()));
        }

        // If this.[[stream]].[[state]] is not "readable", throw a TypeError exception.
        if !self.stream.get().unwrap().is_readable() {
            return Err(Error::Type(c"stream is not readable".to_owned()));
        }

        // Return ? ReadableByteStreamControllerEnqueue(this, chunk).
        self.enqueue(cx, chunk)
    }

    /// <https://streams.spec.whatwg.org/#rbs-controller-error>
    fn Error(&self, cx: &mut JSContext, e: SafeHandleValue) -> Fallible<()> {
        // Perform ! ReadableByteStreamControllerError(this, e).
        self.error(cx, e);
        Ok(())
    }
}