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