Skip to main content

ferrijs_std/stream_web/readable/
byte_controller.rs

1use std::collections::VecDeque;
2
3use crate::utils::{
4    error_messages::ERROR_MSG_ARRAY_BUFFER_DETACHED,
5    option::{Null, Undefined},
6    primordials::{BasePrimordials, Primordial},
7    result::ResultExt,
8};
9use rquickjs::{
10    class::{OwnedBorrow, OwnedBorrowMut, Trace, Tracer},
11    function::Constructor,
12    methods,
13    prelude::{Opt, This},
14    ArrayBuffer, Class, Ctx, Error, Exception, Function, IntoJs, JsLifetime, Object, Promise,
15    Result, TypedArray, Value,
16};
17
18use crate::stream_web::{
19    readable::{
20        byob_reader::{ArrayConstructorPrimordials, ReadableStreamReadIntoRequest, ViewBytes},
21        controller::{
22            ReadableStreamController, ReadableStreamControllerClass, ReadableStreamControllerOwned,
23        },
24        default_controller::ReadableStreamDefaultControllerOwned,
25        default_reader::ReadableStreamReadRequest,
26        objects::{
27            ReadableByteStreamObjects, ReadableStreamBYOBObjects, ReadableStreamClassObjects,
28            ReadableStreamDefaultReaderObjects, ReadableStreamObjects,
29        },
30        reader::ReadableStreamReader,
31        stream::{
32            algorithms::{CancelAlgorithm, PullAlgorithm, StartAlgorithm},
33            source::UnderlyingSource,
34            ReadableStream, ReadableStreamClass, ReadableStreamOwned, ReadableStreamState,
35        },
36    },
37    utils::{
38        class_from_owned_borrow_mut,
39        promise::{promise_resolved_with, upon_promise},
40        UnwrapOrUndefined,
41    },
42};
43
44#[derive(JsLifetime)]
45#[rquickjs::class]
46pub struct ReadableByteStreamController<'js> {
47    auto_allocate_chunk_size: Option<usize>,
48    byob_request: Option<Class<'js, ReadableStreamBYOBRequest<'js>>>,
49    cancel_algorithm: Option<CancelAlgorithm<'js>>,
50    close_requested: bool,
51    pull_again: bool,
52    pull_algorithm: Option<PullAlgorithm<'js>>,
53    pulling: bool,
54    pub(super) pending_pull_intos: VecDeque<PullIntoDescriptor<'js>>,
55    queue: VecDeque<ReadableByteStreamQueueEntry<'js>>,
56    queue_total_size: usize,
57    started: bool,
58    strategy_hwm: f64,
59    pub(super) stream: ReadableStreamClass<'js>,
60
61    pub(super) array_constructor_primordials: ArrayConstructorPrimordials<'js>,
62    constructor_array_buffer: Constructor<'js>,
63    pub(super) function_array_buffer_is_view: Function<'js>,
64}
65
66impl<'js> Trace<'js> for ReadableByteStreamController<'js> {
67    fn trace<'a>(&self, tracer: Tracer<'a, 'js>) {
68        self.auto_allocate_chunk_size.trace(tracer);
69        self.byob_request.trace(tracer);
70        self.cancel_algorithm.trace(tracer);
71        self.pull_algorithm.trace(tracer);
72        self.pending_pull_intos.trace(tracer);
73        self.queue.trace(tracer);
74        self.queue_total_size.trace(tracer);
75        self.started.trace(tracer);
76        self.strategy_hwm.trace(tracer);
77        self.stream.trace(tracer);
78        self.array_constructor_primordials.trace(tracer);
79        self.constructor_array_buffer.trace(tracer);
80        self.function_array_buffer_is_view.trace(tracer);
81    }
82}
83
84pub type ReadableByteStreamControllerClass<'js> = Class<'js, ReadableByteStreamController<'js>>;
85pub(crate) type ReadableByteStreamControllerOwned<'js> =
86    OwnedBorrowMut<'js, ReadableByteStreamController<'js>>;
87
88impl<'js> ReadableByteStreamController<'js> {
89    // SetUpReadableByteStreamControllerFromUnderlyingSource
90    pub(super) fn set_up_readable_byte_stream_controller_from_underlying_source(
91        ctx: &Ctx<'js>,
92        stream: ReadableStreamOwned<'js>,
93        underlying_source: Null<Undefined<Object<'js>>>,
94        underlying_source_dict: UnderlyingSource<'js>,
95        high_water_mark: f64,
96    ) -> Result<()> {
97        let (start_algorithm, pull_algorithm, cancel_algorithm, auto_allocate_chunk_size) = (
98            // If underlyingSourceDict["start"] exists, then set startAlgorithm to an algorithm which returns the result of invoking underlyingSourceDict["start"] with argument list
99            // « controller » and callback this value underlyingSource.
100            underlying_source_dict
101                .start
102                .map(|f| StartAlgorithm::Function {
103                    f,
104                    underlying_source: underlying_source.clone(),
105                })
106                .unwrap_or(StartAlgorithm::ReturnUndefined),
107            // If underlyingSourceDict["pull"] exists, then set pullAlgorithm to an algorithm which returns the result of invoking underlyingSourceDict["pull"] with argument list
108            // « controller » and callback this value underlyingSource.
109            underlying_source_dict
110                .pull
111                .map(|f| PullAlgorithm::Function {
112                    f,
113                    underlying_source: underlying_source.clone(),
114                })
115                .unwrap_or(PullAlgorithm::ReturnPromiseUndefined),
116            // If underlyingSourceDict["cancel"] exists, then set cancelAlgorithm to an algorithm which takes an argument reason and returns the result of invoking underlyingSourceDict["cancel"] with argument list
117            // « reason » and callback this value underlyingSource.
118            underlying_source_dict
119                .cancel
120                .map(|f| CancelAlgorithm::Function {
121                    f,
122                    underlying_source,
123                })
124                .unwrap_or(CancelAlgorithm::ReturnPromiseUndefined),
125            // Let autoAllocateChunkSize be underlyingSourceDict["autoAllocateChunkSize"], if it exists, or undefined otherwise.
126            underlying_source_dict.auto_allocate_chunk_size,
127        );
128
129        // If autoAllocateChunkSize is 0, then throw a TypeError exception.
130        if auto_allocate_chunk_size == Some(0) {
131            return Err(Exception::throw_type(
132                ctx,
133                "autoAllocateChunkSize must be greater than 0",
134            ));
135        }
136
137        Self::set_up_readable_byte_stream_controller(
138            ctx.clone(),
139            stream,
140            start_algorithm,
141            pull_algorithm,
142            cancel_algorithm,
143            high_water_mark,
144            auto_allocate_chunk_size,
145        )?;
146
147        Ok(())
148    }
149
150    pub(super) fn set_up_readable_byte_stream_controller(
151        ctx: Ctx<'js>,
152        stream: ReadableStreamOwned<'js>,
153        start_algorithm: StartAlgorithm<'js>,
154        pull_algorithm: PullAlgorithm<'js>,
155        cancel_algorithm: CancelAlgorithm<'js>,
156        high_water_mark: f64,
157        auto_allocate_chunk_size: Option<usize>,
158    ) -> Result<Class<'js, Self>> {
159        let (stream_class, mut stream) = class_from_owned_borrow_mut(stream);
160
161        let array_constructor_primordials = ArrayConstructorPrimordials::get(&ctx)?.clone();
162        let BasePrimordials {
163            constructor_array_buffer,
164            function_array_buffer_is_view,
165            ..
166        } = &*BasePrimordials::get(&ctx)?;
167
168        let controller = Self {
169            // Set controller.[[stream]] to stream.
170            stream: stream_class,
171
172            // Set controller.[[pullAgain]] and controller.[[pulling]] to false.
173            pull_again: false,
174            pulling: false,
175
176            // Set controller.[[byobRequest]] to null.
177            byob_request: None,
178
179            // Perform ! ResetQueue(controller).
180            queue: VecDeque::new(),
181            queue_total_size: 0,
182
183            // Set controller.[[closeRequested]] and controller.[[started]] to false.
184            close_requested: false,
185            started: false,
186
187            // Set controller.[[strategyHWM]] to highWaterMark.
188            strategy_hwm: high_water_mark,
189
190            // Set controller.[[pullAlgorithm]] to pullAlgorithm.
191            pull_algorithm: Some(pull_algorithm),
192            cancel_algorithm: Some(cancel_algorithm),
193
194            // Set controller.[[autoAllocateChunkSize]] to autoAllocateChunkSize.
195            auto_allocate_chunk_size,
196
197            pending_pull_intos: VecDeque::new(),
198
199            array_constructor_primordials,
200            constructor_array_buffer: constructor_array_buffer.clone(),
201            function_array_buffer_is_view: function_array_buffer_is_view.clone(),
202        };
203
204        let controller_class = Class::instance(ctx.clone(), controller)?;
205
206        // Set stream.[[controller]] to controller.
207        stream.controller =
208            ReadableStreamControllerClass::ReadableStreamByteController(controller_class.clone());
209
210        let objects =
211            ReadableStreamObjects::new_byte(stream, OwnedBorrowMut::from_class(controller_class))
212                .refresh_reader();
213
214        let promise_primordials = objects.stream.promise_primordials.clone();
215
216        // Let startResult be the result of performing startAlgorithm.
217        let (start_result, objects_class) =
218            Self::start_algorithm(ctx.clone(), objects, start_algorithm)?;
219
220        // Let startPromise be a promise resolved with startResult.
221        let start_promise = promise_resolved_with(&ctx, &promise_primordials, Ok(start_result))?;
222
223        let _ = upon_promise::<Value<'js>, _>(ctx.clone(), start_promise, {
224            let objects_class = objects_class.clone();
225            move |ctx, result| {
226                let mut objects =
227                    ReadableStreamObjects::from_class_no_reader(objects_class).refresh_reader();
228                match result {
229                    // Upon fulfillment of startPromise,
230                    Ok(_) => {
231                        // Set controller.[[started]] to true.
232                        objects.controller.started = true;
233                        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
234                        Self::readable_byte_stream_controller_call_pull_if_needed(ctx, objects)?;
235                        Ok(())
236                    },
237                    // Upon rejection of startPromise with reason r,
238                    Err(r) => {
239                        // Perform ! ReadableByteStreamControllerError(controller, r).
240                        Self::readable_byte_stream_controller_error(objects, r)?;
241                        Ok(())
242                    },
243                }
244            }
245        })?;
246
247        Ok(objects_class.controller)
248    }
249
250    fn readable_byte_stream_controller_call_pull_if_needed<R: ReadableStreamReader<'js>>(
251        ctx: Ctx<'js>,
252        objects: ReadableByteStreamObjects<'js, R>,
253    ) -> Result<ReadableByteStreamObjects<'js, R>> {
254        // Let shouldPull be ! ReadableByteStreamControllerShouldCallPull(controller).
255        let (should_pull, mut objects) =
256            Self::readable_byte_stream_controller_should_call_pull(objects);
257
258        // If shouldPull is false, return.
259        if !should_pull {
260            return Ok(objects);
261        }
262
263        // If controller.[[pulling]] is true,
264        if objects.controller.pulling {
265            // Set controller.[[pullAgain]] to true.
266            objects.controller.pull_again = true;
267
268            // Return.
269            return Ok(objects);
270        }
271
272        // Set controller.[[pulling]] to true.
273        objects.controller.pulling = true;
274
275        // Let pullPromise be the result of performing controller.[[pullAlgorithm]].
276        let (pull_promise, objects_class) = Self::pull_algorithm(ctx.clone(), objects)?;
277
278        upon_promise::<Value<'js>, ()>(ctx, pull_promise, {
279            let objects_class = objects_class.clone();
280            move |ctx, result| {
281                let mut objects =
282                    ReadableStreamObjects::from_class_no_reader(objects_class).refresh_reader();
283                match result {
284                    // Upon fulfillment of pullPromise,
285                    Ok(_) => {
286                        // Set controller.[[pulling]] to false.
287                        objects.controller.pulling = false;
288                        // If controller.[[pullAgain]] is true,
289                        if objects.controller.pull_again {
290                            // Set controller.[[pullAgain]] to false.
291                            objects.controller.pull_again = false;
292                            // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
293                            Self::readable_byte_stream_controller_call_pull_if_needed(
294                                ctx, objects,
295                            )?;
296                        };
297                        Ok(())
298                    },
299                    // Upon rejection of pullPromise with reason e,
300                    Err(e) => {
301                        // Perform ! ReadableByteStreamControllerError(controller, e).
302                        Self::readable_byte_stream_controller_error(objects, e)?;
303                        Ok(())
304                    },
305                }
306            }
307        })?;
308
309        Ok(ReadableStreamObjects::from_class(objects_class))
310    }
311
312    fn readable_byte_stream_controller_should_call_pull<R: ReadableStreamReader<'js>>(
313        mut objects: ReadableByteStreamObjects<'js, R>,
314    ) -> (bool, ReadableByteStreamObjects<'js, R>) {
315        // Let stream be controller.[[stream]].
316        match objects.stream.state {
317            ReadableStreamState::Readable => {},
318            // If stream.[[state]] is not "readable", return false.
319            _ => return (false, objects),
320        }
321
322        // If controller.[[closeRequested]] is true, return false.
323        if objects.controller.close_requested {
324            return (false, objects);
325        }
326
327        // If controller.[[started]] is false, return false.
328        if !objects.controller.started {
329            return (false, objects);
330        }
331
332        let (mut has_read_requests, mut has_read_into_requests) = (false, false);
333        objects = objects
334            .with_reader(
335                |objects| {
336                    // If ! ReadableStreamHasDefaultReader(stream) is true and ! ReadableStreamGetNumReadRequests(stream) > 0, return true.
337                    if ReadableStream::readable_stream_get_num_read_requests(&objects.reader) > 0 {
338                        has_read_requests = true;
339                    }
340                    Ok(objects)
341                },
342                |objects| {
343                    // If ! ReadableStreamHasBYOBReader(stream) is true and ! ReadableStreamGetNumReadIntoRequests(stream) > 0, return true.
344                    if ReadableStream::readable_stream_get_num_read_into_requests(&objects.reader)
345                        > 0
346                    {
347                        has_read_into_requests = true;
348                    }
349                    Ok(objects)
350                },
351                Ok,
352            )
353            .unwrap();
354
355        if has_read_requests || has_read_into_requests {
356            return (true, objects);
357        }
358
359        // Let desiredSize be ! ReadableByteStreamControllerGetDesiredSize(controller).
360        let desired_size = objects
361            .controller
362            .readable_byte_stream_controller_get_desired_size(&objects.stream);
363
364        // Assert: desiredSize is not null.
365        if desired_size.0.expect("desired_size must not be null") > 0.0 {
366            // If desiredSize > 0, return true.
367            return (true, objects);
368        }
369
370        // Return false.
371        (false, objects)
372    }
373
374    pub(super) fn readable_byte_stream_controller_error<R: ReadableStreamReader<'js>>(
375        // Let stream be controller.[[stream]].
376        mut objects: ReadableByteStreamObjects<'js, R>,
377        e: Value<'js>,
378    ) -> Result<ReadableByteStreamObjects<'js, R>> {
379        // If stream.[[state]] is not "readable", return.
380        if !matches!(objects.stream.state, ReadableStreamState::Readable) {
381            return Ok(objects);
382        };
383
384        // Perform ! ReadableByteStreamControllerClearPendingPullIntos(controller).
385        objects
386            .controller
387            .readable_byte_stream_controller_clear_pending_pull_intos();
388
389        // Perform ! ResetQueue(controller).
390        objects.controller.reset_queue();
391
392        // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
393        objects
394            .controller
395            .readable_byte_stream_controller_clear_algorithms();
396
397        // Perform ! ReadableStreamError(stream, e).
398        ReadableStream::readable_stream_error(objects, e)
399    }
400
401    fn readable_byte_stream_controller_clear_pending_pull_intos(&mut self) {
402        // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
403        self.readable_byte_stream_controller_invalidate_byob_request();
404
405        // Set controller.[[pendingPullIntos]] to a new empty list.
406        self.pending_pull_intos.clear();
407    }
408
409    fn readable_byte_stream_controller_invalidate_byob_request(&mut self) {
410        let byob_request = match self.byob_request {
411            // If controller.[[byobRequest]] is null, return.
412            None => return,
413            Some(ref byob_request) => byob_request.clone(),
414        };
415        let mut byob_request = OwnedBorrowMut::from_class(byob_request);
416        byob_request.controller = None;
417        byob_request.view = None;
418
419        self.byob_request = None;
420    }
421
422    fn readable_byte_stream_controller_clear_algorithms(&mut self) {
423        self.pull_algorithm = None;
424        self.cancel_algorithm = None;
425    }
426
427    pub(super) fn readable_byte_stream_controller_get_byob_request(
428        ctx: Ctx<'js>,
429        controller: OwnedBorrowMut<'js, Self>,
430    ) -> Result<(
431        Null<Class<'js, ReadableStreamBYOBRequest<'js>>>,
432        OwnedBorrowMut<'js, Self>,
433    )> {
434        // If controller.[[byobRequest]] is null and controller.[[pendingPullIntos]] is not empty,
435        if controller.byob_request.is_none() && !controller.pending_pull_intos.is_empty() {
436            // Let firstDescriptor be controller.[[pendingPullIntos]][0].
437            let first_descriptor = &controller.pending_pull_intos[0];
438
439            // Let view be ! Construct(%Uint8Array%, « firstDescriptor’s buffer, firstDescriptor’s byte offset + firstDescriptor’s bytes filled, firstDescriptor’s byte length − firstDescriptor’s bytes filled »).
440            let view = ViewBytes::from_value(
441                &ctx,
442                &controller.function_array_buffer_is_view,
443                Some(
444                    &controller
445                        .array_constructor_primordials
446                        .constructor_uint8array
447                        .construct((
448                            first_descriptor.buffer.clone(),
449                            first_descriptor.byte_offset + first_descriptor.bytes_filled,
450                            first_descriptor.byte_length - first_descriptor.bytes_filled,
451                        ))?,
452                ),
453            )?;
454
455            let (controller_class, mut controller) = class_from_owned_borrow_mut(controller);
456
457            // Let byobRequest be a new ReadableStreamBYOBRequest.
458            let byob_request = ReadableStreamBYOBRequest {
459                // Set byobRequest.[[controller]] to controller.
460                controller: Some(controller_class),
461                // Set byobRequest.[[view]] to view.
462                view: Some(view),
463            };
464
465            // Set controller.[[byobRequest]] to byobRequest.
466            controller.byob_request = Some(Class::instance(ctx, byob_request)?);
467
468            Ok((Null(controller.byob_request.clone()), controller))
469        } else {
470            // Return controller.[[byobRequest]].
471            Ok((Null(controller.byob_request.clone()), controller))
472        }
473    }
474
475    fn readable_byte_stream_controller_get_desired_size(
476        &self,
477        stream: &ReadableStream<'js>,
478    ) -> Null<f64> {
479        // Let state be controller.[[stream]].[[state]].
480        match stream.state {
481            // If state is "errored", return null.
482            ReadableStreamState::Errored(_) => Null(None),
483            // If state is "closed", return 0.
484            ReadableStreamState::Closed => Null(Some(0.0)),
485            // Return controller.[[strategyHWM]] − controller.[[queueTotalSize]].
486            _ => Null(Some(self.strategy_hwm - self.queue_total_size as f64)),
487        }
488    }
489
490    fn reset_queue(&mut self) {
491        // Set container.[[queue]] to a new empty list.
492        self.queue.clear();
493        // Set container.[[queueTotalSize]] to 0.
494        self.queue_total_size = 0;
495    }
496
497    pub(super) fn readable_byte_stream_controller_close<R: ReadableStreamReader<'js>>(
498        ctx: Ctx<'js>,
499        // Let stream be controller.[[stream]].
500        mut objects: ReadableByteStreamObjects<'js, R>,
501    ) -> Result<ReadableByteStreamObjects<'js, R>> {
502        // If controller.[[closeRequested]] is true or stream.[[state]] is not "readable", return.
503        if objects.controller.close_requested
504            || !matches!(objects.stream.state, ReadableStreamState::Readable)
505        {
506            return Ok(objects);
507        }
508
509        // If controller.[[queueTotalSize]] > 0,
510        if objects.controller.queue_total_size > 0 {
511            // Set controller.[[closeRequested]] to true.
512            objects.controller.close_requested = true;
513            // Return.
514            return Ok(objects);
515        }
516
517        // If controller.[[pendingPullIntos]] is not empty,
518        // Let firstPendingPullInto be controller.[[pendingPullIntos]][0].
519        if let Some(first_pending_pull_into) = objects.controller.pending_pull_intos.front() {
520            // If the remainder after dividing firstPendingPullInto’s bytes filled by firstPendingPullInto’s element size is not 0,
521            if first_pending_pull_into.bytes_filled % first_pending_pull_into.element_size != 0 {
522                // Let e be a new TypeError exception.
523                let e: Value = objects
524                    .stream
525                    .constructor_type_error
526                    .call(("Insufficient bytes to fill elements in the given buffer",))?;
527                Self::readable_byte_stream_controller_error(objects, e.clone())?;
528                return Err(ctx.throw(e));
529            }
530        }
531
532        // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
533        objects
534            .controller
535            .readable_byte_stream_controller_clear_algorithms();
536
537        // Perform ! ReadableStreamClose(stream).
538        ReadableStream::readable_stream_close(ctx, objects)
539    }
540
541    pub(super) fn readable_byte_stream_controller_enqueue<R: ReadableStreamReader<'js>>(
542        ctx: &Ctx<'js>,
543        // Let stream be controller.[[stream]].
544        objects: ReadableByteStreamObjects<'js, R>,
545        chunk: ViewBytes<'js>,
546    ) -> Result<ReadableByteStreamObjects<'js, R>> {
547        Self::readable_byte_stream_controller_enqueue_impl(
548            ctx, objects, chunk, /*skip_transfer=*/ false,
549        )
550    }
551
552    /// Like [`readable_byte_stream_controller_enqueue`] but skips the
553    /// spec-mandated `TransferArrayBuffer(chunk)` step on the incoming
554    /// chunk. Used by producers that already own the backing allocation
555    /// and want to hand it to the stream without QuickJS copying or
556    /// detaching it (e.g. `Blob.stream()`, where the backing
557    /// `ArrayBuffer` must survive multiple `.stream()` calls).
558    ///
559    /// Safety vs. correctness: the chunk we enqueue is NOT detached from
560    /// the producer's perspective, so both producer and consumer see the
561    /// same underlying bytes. This matches the existing non-isolation
562    /// behaviour of `Blob.arrayBuffer()` / `Blob.bytes()`, which already
563    /// return handles that alias the blob's storage. Pending BYOB
564    /// transfers of reader-provided buffers are unaffected — those are
565    /// separate buffers and still use spec-mandated transfer.
566    pub(super) fn readable_byte_stream_controller_enqueue_borrowed<R: ReadableStreamReader<'js>>(
567        ctx: &Ctx<'js>,
568        objects: ReadableByteStreamObjects<'js, R>,
569        chunk: ViewBytes<'js>,
570    ) -> Result<ReadableByteStreamObjects<'js, R>> {
571        Self::readable_byte_stream_controller_enqueue_impl(
572            ctx, objects, chunk, /*skip_transfer=*/ true,
573        )
574    }
575
576    fn readable_byte_stream_controller_enqueue_impl<R: ReadableStreamReader<'js>>(
577        ctx: &Ctx<'js>,
578        mut objects: ReadableByteStreamObjects<'js, R>,
579        chunk: ViewBytes<'js>,
580        skip_transfer: bool,
581    ) -> Result<ReadableByteStreamObjects<'js, R>> {
582        // If controller.[[closeRequested]] is true or stream.[[state]] is not "readable", return.
583        if objects.controller.close_requested
584            || !matches!(objects.stream.state, ReadableStreamState::Readable)
585        {
586            return Ok(objects);
587        };
588
589        // Let buffer be chunk.[[ViewedArrayBuffer]].
590        // Let byteOffset be chunk.[[ByteOffset]].
591        // Let byteLength be chunk.[[ByteLength]].
592        let (buffer, byte_length, byte_offset) = chunk.get_array_buffer()?;
593
594        // If ! IsDetachedBuffer(buffer) is true, throw a TypeError exception.
595        buffer.as_raw().ok_or(Exception::throw_type(
596            ctx,
597            "chunk's buffer is detached and so cannot be enqueued",
598        ))?;
599
600        // Let transferredBuffer be ? TransferArrayBuffer(buffer).
601        // (When `skip_transfer` is true, the caller guarantees that the
602        // buffer is already owned exclusively by the stream for the
603        // purposes of this enqueue — see
604        // `readable_byte_stream_controller_enqueue_borrowed`.)
605        let transferred_buffer = if skip_transfer {
606            buffer
607        } else {
608            transfer_array_buffer(buffer)?
609        };
610
611        // If controller.[[pendingPullIntos]] is not empty,
612        // Let firstPendingPullInto be controller.[[pendingPullIntos]][0].
613        if !objects.controller.pending_pull_intos.is_empty() {
614            // If ! IsDetachedBuffer(firstPendingPullInto’s buffer) is true, throw a TypeError exception.
615            objects.controller.pending_pull_intos[0]
616                    .buffer
617                    .as_raw()
618                    .or_throw_type(
619                        ctx,
620                        "The BYOB request's buffer has been detached and so cannot be filled with an enqueued chunk",
621                    )?;
622
623            // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
624            objects
625                .controller
626                .readable_byte_stream_controller_invalidate_byob_request();
627
628            // Set firstPendingPullInto’s buffer to ! TransferArrayBuffer(firstPendingPullInto’s buffer).
629            objects.controller.pending_pull_intos[0].buffer =
630                transfer_array_buffer(objects.controller.pending_pull_intos[0].buffer.clone())?;
631
632            // If firstPendingPullInto’s reader type is "none", perform ? ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue(controller, firstPendingPullInto).
633            if let PullIntoDescriptorReaderType::None =
634                objects.controller.pending_pull_intos[0].reader_type
635            {
636                objects = Self::readable_byte_stream_enqueue_detached_pull_into_to_queue(
637                    ctx.clone(),
638                    objects,
639                    0,
640                )?;
641            }
642        }
643
644        objects = objects.with_reader(
645            // If ! ReadableStreamHasDefaultReader(stream) is true,
646            |mut objects| {
647                // Perform ! ReadableByteStreamControllerProcessReadRequestsUsingQueue(controller).
648                objects = Self::readable_byte_stream_controller_process_read_requests_using_queue(
649                    objects, ctx,
650                )?;
651
652                // If ! ReadableStreamGetNumReadRequests(stream) is 0,
653                if ReadableStream::readable_stream_get_num_read_requests(&objects.reader) == 0 {
654                    // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue(controller, transferredBuffer, byteOffset, byteLength).
655                    objects
656                        .controller
657                        .readable_byte_stream_controller_enqueue_chunk_to_queue(
658                            transferred_buffer.clone(),
659                            byte_offset,
660                            byte_length,
661                        )
662                } else {
663                    // Otherwise,
664                    // If controller.[[pendingPullIntos]] is not empty,
665                    if !objects.controller.pending_pull_intos.is_empty() {
666                        // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
667                        objects
668                            .controller
669                            .readable_byte_stream_controller_shift_pending_pull_into();
670                    }
671
672                    // Let transferredView be ! Construct(%Uint8Array%, « transferredBuffer, byteOffset, byteLength »).
673                    let transferred_view = ViewBytes::from_value(
674                        ctx,
675                        &objects.controller.function_array_buffer_is_view,
676                        Some(
677                            &objects
678                                .controller
679                                .array_constructor_primordials
680                                .constructor_uint8array
681                                .construct((
682                                    transferred_buffer.clone(),
683                                    byte_offset,
684                                    byte_length,
685                                ))?,
686                        ),
687                    );
688
689                    // Perform ! ReadableStreamFulfillReadRequest(stream, transferredView, false).
690                    objects = ReadableStream::readable_stream_fulfill_read_request(
691                        ctx,
692                        objects,
693                        transferred_view.into_js(ctx)?,
694                        false,
695                    )?;
696                }
697
698                Ok(objects)
699            },
700            |mut objects| {
701                // Otherwise, if ! ReadableStreamHasBYOBReader(stream) is true,
702                // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue(controller, transferredBuffer, byteOffset, byteLength).
703                objects
704                    .controller
705                    .readable_byte_stream_controller_enqueue_chunk_to_queue(
706                        transferred_buffer.clone(),
707                        byte_offset,
708                        byte_length,
709                    );
710                // Perform ! ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
711
712                Self::readable_byte_stream_controller_process_pull_into_descriptors_using_queue(
713                    ctx, objects,
714                )
715            },
716            |mut objects| {
717                // Otherwise,
718                // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue(controller, transferredBuffer, byteOffset, byteLength).
719                objects
720                    .controller
721                    .readable_byte_stream_controller_enqueue_chunk_to_queue(
722                        transferred_buffer.clone(),
723                        byte_offset,
724                        byte_length,
725                    );
726
727                Ok(objects)
728            },
729        )?;
730
731        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
732        Self::readable_byte_stream_controller_call_pull_if_needed(ctx.clone(), objects)
733    }
734
735    fn readable_byte_stream_enqueue_detached_pull_into_to_queue<R: ReadableStreamReader<'js>>(
736        ctx: Ctx<'js>,
737        mut objects: ReadableByteStreamObjects<'js, R>,
738        pull_into_descriptor_index: usize,
739    ) -> Result<ReadableByteStreamObjects<'js, R>> {
740        let pull_into_descriptor =
741            &objects.controller.pending_pull_intos[pull_into_descriptor_index];
742        // If pullIntoDescriptor’s bytes filled > 0, perform ? ReadableByteStreamControllerEnqueueClonedChunkToQueue(controller, pullIntoDescriptor’s buffer, pullIntoDescriptor’s byte offset, pullIntoDescriptor’s bytes filled).
743        if pull_into_descriptor.bytes_filled > 0 {
744            let buffer = pull_into_descriptor.buffer.clone();
745            let byte_offset = pull_into_descriptor.byte_offset;
746            let bytes_filled = pull_into_descriptor.bytes_filled;
747            objects = Self::readable_byte_stream_controller_enqueue_cloned_chunk_to_queue(
748                ctx,
749                objects,
750                &buffer,
751                byte_offset,
752                bytes_filled,
753            )?;
754        }
755
756        // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
757        objects
758            .controller
759            .readable_byte_stream_controller_shift_pending_pull_into();
760
761        Ok(objects)
762    }
763
764    fn readable_byte_stream_controller_process_read_requests_using_queue(
765        mut objects: ReadableStreamDefaultReaderObjects<'js, OwnedBorrowMut<'js, Self>>,
766        ctx: &Ctx<'js>,
767    ) -> Result<ReadableStreamDefaultReaderObjects<'js, OwnedBorrowMut<'js, Self>>> {
768        // While reader.[[readRequests]] is not empty,
769        while !objects.reader.read_requests.is_empty() {
770            // If controller.[[queueTotalSize]] is 0, return.
771            if objects.controller.queue_total_size == 0 {
772                return Ok(objects);
773            }
774
775            // Let readRequest be reader.[[readRequests]][0].
776            // Remove readRequest from reader.[[readRequests]].
777            let read_request = objects.reader.read_requests.pop_front().unwrap();
778            // Perform ! ReadableByteStreamControllerFillReadRequestFromQueue(controller, readRequest).
779            objects = Self::readable_byte_stream_controller_fill_read_request_from_queue(
780                ctx,
781                objects,
782                read_request,
783            )?;
784        }
785
786        Ok(objects)
787    }
788
789    fn readable_byte_stream_controller_shift_pending_pull_into(
790        &mut self,
791    ) -> PullIntoDescriptor<'js> {
792        // Invalidate byobRequest since the first pending pull-into is being removed
793        self.readable_byte_stream_controller_invalidate_byob_request();
794        // Let descriptor be controller.[[pendingPullIntos]][0].
795        // Remove descriptor from controller.[[pendingPullIntos]].
796        // Return descriptor.
797        self.pending_pull_intos.pop_front().expect(
798            "ReadableByteStreamControllerShiftPendingPullInto called on empty pendingPullIntos",
799        )
800    }
801
802    fn readable_byte_stream_controller_enqueue_chunk_to_queue(
803        &mut self,
804        buffer: ArrayBuffer<'js>,
805        byte_offset: usize,
806        byte_length: usize,
807    ) {
808        // Append a new readable byte stream queue entry with buffer buffer, byte offset byteOffset, and byte length byteLength to controller.[[queue]].
809        self.queue.push_back(ReadableByteStreamQueueEntry {
810            buffer,
811            byte_offset,
812            byte_length,
813        });
814
815        // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] + byteLength.
816        self.queue_total_size += byte_length;
817    }
818
819    fn readable_byte_stream_controller_process_pull_into_descriptors_using_queue<
820        R: ReadableStreamReader<'js>,
821    >(
822        ctx: &Ctx<'js>,
823        mut objects: ReadableByteStreamObjects<'js, R>,
824    ) -> Result<ReadableByteStreamObjects<'js, R>> {
825        // While controller.[[pendingPullIntos]] is not empty,
826        while !objects.controller.pending_pull_intos.is_empty() {
827            // If controller.[[queueTotalSize]] is 0, return.
828            if objects.controller.queue_total_size == 0 {
829                return Ok(objects);
830            }
831
832            // Let pullIntoDescriptor be controller.[[pendingPullIntos]][0].
833            let mut pull_into_descriptor_ref = PullIntoDescriptorRefMut::Index(0);
834
835            // If ! ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller, pullIntoDescriptor) is true,
836            if objects
837                .controller
838                .readable_byte_stream_controller_fill_pull_into_descriptor_from_queue(
839                    ctx,
840                    &mut pull_into_descriptor_ref,
841                )?
842            {
843                // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
844                let pull_into_descriptor = objects
845                    .controller
846                    .readable_byte_stream_controller_shift_pending_pull_into();
847
848                // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]], pullIntoDescriptor).
849                objects = Self::readable_byte_stream_controller_commit_pull_into_descriptor(
850                    ctx.clone(),
851                    objects,
852                    pull_into_descriptor,
853                )?;
854            }
855        }
856        Ok(objects)
857    }
858
859    fn readable_byte_stream_controller_enqueue_cloned_chunk_to_queue<
860        R: ReadableStreamReader<'js>,
861    >(
862        ctx: Ctx<'js>,
863        mut objects: ReadableByteStreamObjects<'js, R>,
864        buffer: &ArrayBuffer<'js>,
865        byte_offset: usize,
866        byte_length: usize,
867    ) -> Result<ReadableByteStreamObjects<'js, R>> {
868        // Let cloneResult be CloneArrayBuffer(buffer, byteOffset, byteLength, %ArrayBuffer%).
869        let clone_result = match ArrayBuffer::new_copy(
870            ctx.clone(),
871            // SAFETY: `new_copy` cannot run user script.
872            &unsafe { buffer.as_bytes() }.expect(
873                "ReadableByteStreamControllerEnqueueClonedChunkToQueue called on detached buffer",
874            )[byte_offset..byte_offset + byte_length],
875        ) {
876            Ok(clone_result) => clone_result,
877            Err(Error::Exception) => {
878                let err = ctx.catch();
879                Self::readable_byte_stream_controller_error(objects, err.clone())?;
880                return Err(ctx.throw(err));
881            },
882            Err(err) => return Err(err),
883        };
884
885        // Perform ! ReadableByteStreamControllerEnqueueChunkToQueue(controller, cloneResult.[[Value]], 0, byteLength).
886        objects
887            .controller
888            .readable_byte_stream_controller_enqueue_chunk_to_queue(clone_result, 0, byte_length);
889
890        Ok(objects)
891    }
892
893    fn readable_byte_stream_controller_fill_read_request_from_queue(
894        ctx: &Ctx<'js>,
895        mut objects: ReadableStreamDefaultReaderObjects<'js, OwnedBorrowMut<'js, Self>>,
896        read_request: impl ReadableStreamReadRequest<'js>,
897    ) -> Result<ReadableStreamDefaultReaderObjects<'js, OwnedBorrowMut<'js, Self>>> {
898        let entry = {
899            // Assert: controller.[[queueTotalSize]] > 0.
900            // Let entry be controller.[[queue]][0].
901            // Remove entry from controller.[[queue]].
902            let entry = objects.controller.queue.pop_front().expect(
903                "ReadableByteStreamControllerFillReadRequestFromQueue called with empty queue",
904            );
905
906            // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] − entry’s byte length.
907            objects.controller.queue_total_size -= entry.byte_length;
908
909            entry
910        };
911
912        // Perform ! ReadableByteStreamControllerHandleQueueDrain(controller).
913        objects = Self::readable_byte_stream_controller_handle_queue_drain(ctx.clone(), objects)?;
914
915        // Let view be ! Construct(%Uint8Array%, « entry’s buffer, entry’s byte offset, entry’s byte length »).
916        let view: TypedArray<u8> = objects
917            .controller
918            .array_constructor_primordials
919            .constructor_uint8array
920            .construct((entry.buffer, entry.byte_offset, entry.byte_length))?;
921
922        // Perform readRequest’s chunk steps, given view.
923        read_request.chunk_steps_typed(objects, view.into_value())
924    }
925
926    fn readable_byte_stream_controller_fill_pull_into_descriptor_from_queue<'a>(
927        &'a mut self,
928        ctx: &Ctx<'js>,
929        pull_into_descriptor_ref: &mut PullIntoDescriptorRefMut<'js, 'a>,
930    ) -> Result<bool> {
931        let (mut total_bytes_to_copy_remaining, ready) = {
932            let pull_into_descriptor = match pull_into_descriptor_ref {
933                PullIntoDescriptorRefMut::Index(i) => &mut self.pending_pull_intos[*i],
934                PullIntoDescriptorRefMut::Owned(r) => r,
935            };
936            // Let maxBytesToCopy be min(controller.[[queueTotalSize]], pullIntoDescriptor’s byte length − pullIntoDescriptor’s bytes filled).
937            let max_bytes_to_copy: usize = std::cmp::min(
938                self.queue_total_size,
939                pull_into_descriptor.byte_length - pull_into_descriptor.bytes_filled,
940            );
941
942            // Let maxBytesFilled be pullIntoDescriptor’s bytes filled + maxBytesToCopy.
943            let max_bytes_filled = pull_into_descriptor.bytes_filled + max_bytes_to_copy;
944
945            // Let totalBytesToCopyRemaining be maxBytesToCopy.
946            let mut total_bytes_to_copy_remaining = max_bytes_to_copy;
947
948            // Let ready be false.
949            let mut ready = false;
950
951            // Let remainderBytes be the remainder after dividing maxBytesFilled by pullIntoDescriptor’s element size.
952            let remainder_bytes = max_bytes_filled % pull_into_descriptor.element_size;
953
954            // Let maxAlignedBytes be maxBytesFilled − remainderBytes.
955            let max_aligned_bytes = max_bytes_filled - remainder_bytes;
956
957            // If maxAlignedBytes ≥ pullIntoDescriptor’s minimum fill,
958            if max_aligned_bytes >= pull_into_descriptor.minimum_fill {
959                // Set totalBytesToCopyRemaining to maxAlignedBytes − pullIntoDescriptor’s bytes filled.
960                total_bytes_to_copy_remaining =
961                    max_aligned_bytes - pull_into_descriptor.bytes_filled;
962                // Set ready to true.
963                ready = true
964            }
965
966            (total_bytes_to_copy_remaining, ready)
967        };
968
969        // Let queue be controller.[[queue]].
970        // While totalBytesToCopyRemaining > 0,
971        while total_bytes_to_copy_remaining > 0 {
972            let bytes_to_copy = {
973                let pull_into_descriptor = match pull_into_descriptor_ref {
974                    PullIntoDescriptorRefMut::Index(i) => &mut self.pending_pull_intos[*i],
975                    PullIntoDescriptorRefMut::Owned(r) => r,
976                };
977
978                // Let headOfQueue be queue[0].
979                let head_of_queue = self
980                    .queue
981                    .front_mut()
982                    .expect("empty queue with bytes to copy");
983                // Let bytesToCopy be min(totalBytesToCopyRemaining, headOfQueue’s byte length).
984                let bytes_to_copy: usize =
985                    std::cmp::min(total_bytes_to_copy_remaining, head_of_queue.byte_length);
986                // Let destStart be pullIntoDescriptor’s byte offset + pullIntoDescriptor’s bytes filled.
987                let dest_start: usize =
988                    pull_into_descriptor.byte_offset + pull_into_descriptor.bytes_filled;
989                // Perform ! CopyDataBlockBytes(pullIntoDescriptor’s buffer.[[ArrayBufferData]], destStart, headOfQueue’s buffer.[[ArrayBufferData]], headOfQueue’s byte offset, bytesToCopy).
990                copy_data_block_bytes(
991                    ctx,
992                    &pull_into_descriptor.buffer,
993                    dest_start,
994                    &head_of_queue.buffer,
995                    head_of_queue.byte_offset,
996                    bytes_to_copy,
997                )?;
998                if head_of_queue.byte_length == bytes_to_copy {
999                    // If headOfQueue’s byte length is bytesToCopy,
1000                    // Remove queue[0].
1001                    self.queue.pop_front();
1002                } else {
1003                    // Otherwise,
1004                    // Set headOfQueue’s byte offset to headOfQueue’s byte offset + bytesToCopy.
1005                    head_of_queue.byte_offset += bytes_to_copy;
1006                    // Set headOfQueue’s byte length to headOfQueue’s byte length − bytesToCopy.
1007                    head_of_queue.byte_length -= bytes_to_copy
1008                }
1009
1010                // Set controller.[[queueTotalSize]] to controller.[[queueTotalSize]] − bytesToCopy.
1011                self.queue_total_size -= bytes_to_copy;
1012
1013                bytes_to_copy
1014            };
1015
1016            // Perform ! ReadableByteStreamControllerFillHeadPullIntoDescriptor(controller, bytesToCopy, pullIntoDescriptor).
1017            self.readable_byte_stream_controller_fill_head_pull_into_descriptor(
1018                bytes_to_copy,
1019                pull_into_descriptor_ref,
1020            );
1021
1022            // Set totalBytesToCopyRemaining to totalBytesToCopyRemaining − bytesToCopy.
1023            total_bytes_to_copy_remaining -= bytes_to_copy
1024        }
1025
1026        Ok(ready)
1027    }
1028
1029    fn readable_byte_stream_controller_commit_pull_into_descriptor<R: ReadableStreamReader<'js>>(
1030        ctx: Ctx<'js>,
1031        objects: ReadableByteStreamObjects<'js, R>,
1032        pull_into_descriptor: PullIntoDescriptor<'js>,
1033    ) -> Result<ReadableByteStreamObjects<'js, R>> {
1034        // Let done be false.
1035        let mut done = false;
1036        // If stream.[[state]] is "closed",
1037        if matches!(objects.stream.state, ReadableStreamState::Closed) {
1038            // Set done to true.
1039            done = true
1040        }
1041
1042        let reader_type = pull_into_descriptor.reader_type;
1043
1044        // Let filledView be ! ReadableByteStreamControllerConvertPullIntoDescriptor(pullIntoDescriptor).
1045        let filled_view = Self::readable_byte_stream_controller_convert_pull_into_descriptor(
1046            ctx.clone(),
1047            &objects.stream.function_array_buffer_is_view,
1048            pull_into_descriptor,
1049        )?;
1050
1051        if let PullIntoDescriptorReaderType::Default = reader_type {
1052            // If pullIntoDescriptor’s reader type is "default",
1053            objects.with_assert_default_reader(|objects| {
1054                // Perform ! ReadableStreamFulfillReadRequest(stream, filledView, done).
1055                ReadableStream::readable_stream_fulfill_read_request(
1056                    &ctx,
1057                    objects,
1058                    filled_view.into_js(&ctx)?,
1059                    done,
1060                )
1061            })
1062        } else {
1063            // Otherwise,
1064            objects.with_assert_byob_reader(|objects| {
1065                // Perform ! ReadableStreamFulfillReadIntoRequest(stream, filledView, done).
1066                ReadableStream::readable_stream_fulfill_read_into_request(
1067                    &ctx,
1068                    objects,
1069                    filled_view,
1070                    done,
1071                )
1072            })
1073        }
1074    }
1075
1076    fn readable_byte_stream_controller_handle_queue_drain<R: ReadableStreamReader<'js>>(
1077        ctx: Ctx<'js>,
1078        mut objects: ReadableByteStreamObjects<'js, R>,
1079    ) -> Result<ReadableByteStreamObjects<'js, R>> {
1080        // If controller.[[queueTotalSize]] is 0 and controller.[[closeRequested]] is true,
1081        if objects.controller.queue_total_size == 0 && objects.controller.close_requested {
1082            // Perform ! ReadableByteStreamControllerClearAlgorithms(controller).
1083            objects
1084                .controller
1085                .readable_byte_stream_controller_clear_algorithms();
1086            // Perform ! ReadableStreamClose(controller.[[stream]]).
1087            ReadableStream::readable_stream_close(ctx, objects)
1088        } else {
1089            // Otherwise,
1090            // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
1091            Self::readable_byte_stream_controller_call_pull_if_needed(ctx.clone(), objects)
1092        }
1093    }
1094
1095    fn readable_byte_stream_controller_convert_pull_into_descriptor(
1096        ctx: Ctx<'js>,
1097        function_array_buffer_is_view: &Function<'js>,
1098        pull_into_descriptor: PullIntoDescriptor<'js>,
1099    ) -> Result<ViewBytes<'js>> {
1100        let PullIntoDescriptor {
1101            // Let bytesFilled be pullIntoDescriptor’s bytes filled.
1102            bytes_filled,
1103            // Let elementSize be pullIntoDescriptor’s element size.
1104            element_size,
1105            byte_offset,
1106            buffer,
1107            ..
1108        } = pull_into_descriptor;
1109        // Let buffer be ! TransferArrayBuffer(pullIntoDescriptor’s buffer).
1110        let buffer = transfer_array_buffer(buffer);
1111        // Return ! Construct(pullIntoDescriptor’s view constructor, « buffer, pullIntoDescriptor’s byte offset, bytesFilled ÷ elementSize »).
1112        let view: Object = pull_into_descriptor.view_constructor.construct((
1113            buffer,
1114            byte_offset,
1115            bytes_filled / element_size,
1116        ))?;
1117        ViewBytes::from_object(&ctx, function_array_buffer_is_view, &view)
1118    }
1119
1120    pub(super) fn readable_byte_stream_controller_pull_into(
1121        ctx: &Ctx<'js>,
1122        // Let stream be controller.[[stream]].
1123        mut objects: ReadableStreamBYOBObjects<'js>,
1124        view: ViewBytes<'js>,
1125        min: u64,
1126        read_into_request: impl ReadableStreamReadIntoRequest<'js> + 'js,
1127    ) -> Result<ReadableStreamBYOBObjects<'js>> {
1128        // Set elementSize to the element size specified in the typed array constructors table for view.[[TypedArrayName]].
1129        // Set ctor to the constructor specified in the typed array constructors table for view.[[TypedArrayName]].
1130        let (element_size, ctor) = (
1131            view.element_size(),
1132            objects
1133                .controller
1134                .array_constructor_primordials
1135                .for_view_bytes(&view),
1136        );
1137
1138        // Let minimumFill be min × elementSize.
1139        let minimum_fill: usize = (min as usize) * element_size;
1140
1141        // Let byteOffset be view.[[ByteOffset]].
1142        // Let byteLength be view.[[ByteLength]].
1143        let (buffer, byte_length, byte_offset) = view.get_array_buffer()?;
1144
1145        // Let bufferResult be TransferArrayBuffer(view.[[ViewedArrayBuffer]]).
1146        let buffer_result = transfer_array_buffer(buffer);
1147        let buffer = match buffer_result {
1148            // If bufferResult is an abrupt completion,
1149            Err(Error::Exception) => {
1150                // Perform readIntoRequest’s error steps, given bufferResult.[[Value]].
1151                objects = read_into_request.error_steps(objects, ctx.catch())?;
1152                // Return.
1153                return Ok(objects);
1154            },
1155            Err(err) => return Err(err),
1156            // Let buffer be bufferResult.[[Value]].
1157            Ok(buffer) => buffer,
1158        };
1159
1160        let buffer_byte_length = buffer.len();
1161        // Let pullIntoDescriptor be a new pull-into descriptor with
1162        let mut pull_into_descriptor = PullIntoDescriptor {
1163            buffer,
1164            buffer_byte_length,
1165            byte_offset,
1166            byte_length,
1167            bytes_filled: 0,
1168            minimum_fill,
1169            element_size,
1170            view_constructor: ctor.clone(),
1171            reader_type: PullIntoDescriptorReaderType::Byob,
1172        };
1173
1174        // If controller.[[pendingPullIntos]] is not empty,
1175        if !objects.controller.pending_pull_intos.is_empty() {
1176            // Append pullIntoDescriptor to controller.[[pendingPullIntos]].
1177            objects
1178                .controller
1179                .pending_pull_intos
1180                .push_back(pull_into_descriptor);
1181
1182            // Perform ! ReadableStreamAddReadIntoRequest(stream, readIntoRequest).
1183            ReadableStream::readable_stream_add_read_into_request(
1184                &mut objects.reader,
1185                read_into_request,
1186            );
1187
1188            // Return.
1189            return Ok(objects);
1190        }
1191
1192        // If stream.[[state]] is "closed",
1193        if matches!(objects.stream.state, ReadableStreamState::Closed) {
1194            // Let emptyView be ! Construct(ctor, « pullIntoDescriptor’s buffer, pullIntoDescriptor’s byte offset, 0 »).
1195            let empty_view: Value<'js> = ctor.construct((
1196                pull_into_descriptor.buffer,
1197                pull_into_descriptor.byte_offset,
1198                0,
1199            ))?;
1200
1201            // Perform readIntoRequest’s close steps, given emptyView.
1202            objects = read_into_request.close_steps(objects, empty_view)?;
1203
1204            // Return.
1205            return Ok(objects);
1206        }
1207
1208        // If controller.[[queueTotalSize]] > 0,
1209        if objects.controller.queue_total_size > 0 {
1210            // If ! ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller, pullIntoDescriptor) is true,
1211            if objects
1212                .controller
1213                .readable_byte_stream_controller_fill_pull_into_descriptor_from_queue(
1214                    ctx,
1215                    &mut PullIntoDescriptorRefMut::Owned(&mut pull_into_descriptor),
1216                )?
1217            {
1218                // Let filledView be ! ReadableByteStreamControllerConvertPullIntoDescriptor(pullIntoDescriptor).
1219                let filled_view = objects
1220                    .controller
1221                    .readable_byte_steam_controller_convert_pull_into_descriptor(
1222                        pull_into_descriptor,
1223                    )?;
1224
1225                // Perform ! ReadableByteStreamControllerHandleQueueDrain(controller).
1226                objects =
1227                    Self::readable_byte_stream_controller_handle_queue_drain(ctx.clone(), objects)?;
1228
1229                // Perform readIntoRequest’s chunk steps, given filledView.
1230                // Return.
1231                return read_into_request.chunk_steps(objects, filled_view);
1232            }
1233
1234            // If controller.[[closeRequested]] is true,
1235            if objects.controller.close_requested {
1236                // Let e be a TypeError exception.
1237                let e: Value = objects
1238                    .stream
1239                    .constructor_type_error
1240                    .call(("Insufficient bytes to fill elements in the given buffer",))?;
1241
1242                // Perform ! ReadableByteStreamControllerError(controller, e).
1243                objects = Self::readable_byte_stream_controller_error(objects, e.clone())?;
1244
1245                // Perform readIntoRequest’s error steps, given e.
1246                // Return.
1247                return read_into_request.error_steps(objects, e);
1248            }
1249        }
1250
1251        // Append pullIntoDescriptor to controller.[[pendingPullIntos]].
1252        objects
1253            .controller
1254            .pending_pull_intos
1255            .push_back(pull_into_descriptor);
1256
1257        // Perform ! ReadableStreamAddReadIntoRequest(stream, readIntoRequest).
1258        ReadableStream::readable_stream_add_read_into_request(
1259            &mut objects.reader,
1260            read_into_request,
1261        );
1262
1263        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(controller).
1264        Self::readable_byte_stream_controller_call_pull_if_needed(ctx.clone(), objects)
1265    }
1266
1267    fn readable_byte_steam_controller_convert_pull_into_descriptor(
1268        &mut self,
1269        pull_into_descriptor: PullIntoDescriptor<'js>,
1270    ) -> Result<Value<'js>> {
1271        // Let bytesFilled be pullIntoDescriptor’s bytes filled.
1272        let bytes_filled = pull_into_descriptor.bytes_filled;
1273
1274        // Let elementSize be pullIntoDescriptor’s element size.
1275        let element_size = pull_into_descriptor.element_size;
1276
1277        // Let buffer be ! TransferArrayBuffer(pullIntoDescriptor’s buffer).
1278        let buffer = transfer_array_buffer(pull_into_descriptor.buffer)?;
1279
1280        // Return ! Construct(pullIntoDescriptor’s view constructor, « buffer, pullIntoDescriptor’s byte offset, bytesFilled ÷ elementSize »).
1281        pull_into_descriptor.view_constructor.construct((
1282            buffer,
1283            pull_into_descriptor.byte_offset,
1284            bytes_filled / element_size,
1285        ))
1286    }
1287
1288    pub(super) fn readable_byte_stream_controller_respond<R: ReadableStreamReader<'js>>(
1289        ctx: Ctx<'js>,
1290        mut objects: ReadableByteStreamObjects<'js, R>,
1291        bytes_written: usize,
1292    ) -> Result<()> {
1293        // Let firstDescriptor be controller.[[pendingPullIntos]][0].
1294        let first_descriptor = &mut objects.controller.pending_pull_intos[0];
1295
1296        // Let state be controller.[[stream]].[[state]].
1297        match objects.stream.state {
1298            // If state is "closed",
1299            ReadableStreamState::Closed => {
1300                // If bytesWritten is not 0, throw a TypeError exception.
1301                if bytes_written != 0 {
1302                    return Err(Exception::throw_type(
1303                        &ctx,
1304                        "bytesWritten must be 0 when calling respond() on a closed stream",
1305                    ));
1306                }
1307            },
1308            // Otherwise,
1309            _ => {
1310                // If bytesWritten is 0, throw a TypeError exception.
1311                if bytes_written == 0 {
1312                    return Err(Exception::throw_type(
1313                        &ctx,
1314                        "bytesWritten must be greater than 0 when calling respond() on a readable stream",
1315                    ));
1316                }
1317
1318                // If firstDescriptor’s bytes filled + bytesWritten > firstDescriptor’s byte length, throw a RangeError exception.
1319                if first_descriptor.bytes_filled + bytes_written > first_descriptor.byte_length {
1320                    return Err(Exception::throw_range(&ctx, "bytesWritten out of range'"));
1321                }
1322            },
1323        };
1324
1325        // Set firstDescriptor’s buffer to ! TransferArrayBuffer(firstDescriptor’s buffer).
1326        first_descriptor.buffer = transfer_array_buffer(first_descriptor.buffer.clone())?;
1327
1328        // Perform ? ReadableByteStreamControllerRespondInternal(controller, bytesWritten).
1329        Self::readable_byte_stream_controller_respond_internal(ctx, objects, bytes_written)
1330    }
1331
1332    fn readable_byte_stream_controller_respond_internal<R: ReadableStreamReader<'js>>(
1333        ctx: Ctx<'js>,
1334        mut objects: ReadableByteStreamObjects<'js, R>,
1335        bytes_written: usize,
1336    ) -> Result<()> {
1337        // Let firstDescriptor be controller.[[pendingPullIntos]][0].
1338        let first_descriptor_index = 0;
1339
1340        // Perform ! ReadableByteStreamControllerInvalidateBYOBRequest(controller).
1341        objects
1342            .controller
1343            .readable_byte_stream_controller_invalidate_byob_request();
1344
1345        // Let state be controller.[[stream]].[[state]].
1346        match objects.stream.state {
1347            // If state is "closed",
1348            ReadableStreamState::Closed => {
1349                // Perform ! ReadableByteStreamControllerRespondInClosedState(controller, firstDescriptor).
1350                objects = Self::readable_byte_stream_controller_respond_in_closed_state(
1351                    ctx.clone(),
1352                    objects,
1353                    first_descriptor_index,
1354                )?;
1355            },
1356            // Otherwise
1357            _ => {
1358                // Perform ? ReadableByteStreamControllerRespondInReadableState(controller, bytesWritten, firstDescriptor).
1359                objects = Self::readable_byte_stream_controller_respond_in_readable_state(
1360                    ctx.clone(),
1361                    objects,
1362                    bytes_written,
1363                    first_descriptor_index,
1364                )?
1365            },
1366        };
1367
1368        _ = Self::readable_byte_stream_controller_call_pull_if_needed(ctx, objects)?;
1369        Ok(())
1370    }
1371
1372    fn readable_byte_stream_controller_respond_in_closed_state<R: ReadableStreamReader<'js>>(
1373        ctx: Ctx<'js>,
1374        // Let stream be controller.[[stream]].
1375        mut objects: ReadableByteStreamObjects<'js, R>,
1376        first_descriptor_index: usize,
1377    ) -> Result<ReadableByteStreamObjects<'js, R>> {
1378        // If firstDescriptor’s reader type is "none", perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
1379        if let PullIntoDescriptorReaderType::None =
1380            objects.controller.pending_pull_intos[first_descriptor_index].reader_type
1381        {
1382            objects
1383                .controller
1384                .readable_byte_stream_controller_shift_pending_pull_into();
1385        }
1386
1387        // If ! ReadableStreamHasBYOBReader(stream) is true,
1388        objects.with_reader(
1389            Ok,
1390            |mut objects| {
1391                // While ! ReadableStreamGetNumReadIntoRequests(stream) > 0,
1392                while ReadableStream::readable_stream_get_num_read_into_requests(&objects.reader)
1393                    > 0
1394                {
1395                    // Let pullIntoDescriptor be ! ReadableByteStreamControllerShiftPendingPullInto(controller).
1396                    let pull_into_descriptor = objects
1397                        .controller
1398                        .readable_byte_stream_controller_shift_pending_pull_into();
1399
1400                    // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(stream, pullIntoDescriptor).
1401                    objects = Self::readable_byte_stream_controller_commit_pull_into_descriptor(
1402                        ctx.clone(),
1403                        objects,
1404                        pull_into_descriptor,
1405                    )?;
1406                }
1407
1408                Ok(objects)
1409            },
1410            Ok,
1411        )
1412    }
1413
1414    fn readable_byte_stream_controller_respond_in_readable_state<R: ReadableStreamReader<'js>>(
1415        ctx: Ctx<'js>,
1416        // Let stream be controller.[[stream]].
1417        mut objects: ReadableByteStreamObjects<'js, R>,
1418        bytes_written: usize,
1419        pull_into_descriptor_index: usize,
1420    ) -> Result<ReadableByteStreamObjects<'js, R>> {
1421        // Perform ! ReadableByteStreamControllerFillHeadPullIntoDescriptor(controller, bytesWritten, pullIntoDescriptor).
1422        objects
1423            .controller
1424            .readable_byte_stream_controller_fill_head_pull_into_descriptor(
1425                bytes_written,
1426                &mut PullIntoDescriptorRefMut::Index(pull_into_descriptor_index),
1427            );
1428
1429        // If pullIntoDescriptor’s reader type is "none",
1430        if let PullIntoDescriptorReaderType::None =
1431            objects.controller.pending_pull_intos[pull_into_descriptor_index].reader_type
1432        {
1433            // Perform ? ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue(controller, pullIntoDescriptor).
1434            objects = Self::readable_byte_stream_enqueue_detached_pull_into_to_queue(
1435                ctx.clone(),
1436                objects,
1437                pull_into_descriptor_index,
1438            )?;
1439            // Perform ! ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
1440            // Return.
1441            return Self::readable_byte_stream_controller_process_pull_into_descriptors_using_queue(
1442                &ctx, objects,
1443            );
1444        }
1445
1446        // If pullIntoDescriptor’s bytes filled < pullIntoDescriptor’s minimum fill, return.
1447        if objects.controller.pending_pull_intos[pull_into_descriptor_index].bytes_filled
1448            < objects.controller.pending_pull_intos[pull_into_descriptor_index].minimum_fill
1449        {
1450            return Ok(objects);
1451        }
1452
1453        // Perform ! ReadableByteStreamControllerShiftPendingPullInto(controller).
1454        let mut pull_into_descriptor = objects
1455            .controller
1456            .readable_byte_stream_controller_shift_pending_pull_into();
1457
1458        // Let remainderSize be the remainder after dividing pullIntoDescriptor’s bytes filled by pullIntoDescriptor’s element size.
1459        let remainder_size = pull_into_descriptor.bytes_filled % pull_into_descriptor.element_size;
1460
1461        // If remainderSize > 0,
1462        if remainder_size > 0 {
1463            // Let end be pullIntoDescriptor’s byte offset + pullIntoDescriptor’s bytes filled.
1464            let end = pull_into_descriptor.byte_offset + pull_into_descriptor.bytes_filled;
1465
1466            let buffer = pull_into_descriptor.buffer.clone();
1467
1468            // Perform ? ReadableByteStreamControllerEnqueueClonedChunkToQueue(controller, pullIntoDescriptor’s buffer, end − remainderSize, remainderSize).
1469            objects = Self::readable_byte_stream_controller_enqueue_cloned_chunk_to_queue(
1470                ctx.clone(),
1471                objects,
1472                &buffer,
1473                end - remainder_size,
1474                remainder_size,
1475            )?;
1476        }
1477
1478        // Set pullIntoDescriptor’s bytes filled to pullIntoDescriptor’s bytes filled − remainderSize.
1479        pull_into_descriptor.bytes_filled -= remainder_size;
1480
1481        // Perform ! ReadableByteStreamControllerCommitPullIntoDescriptor(controller.[[stream]], pullIntoDescriptor).
1482        objects = Self::readable_byte_stream_controller_commit_pull_into_descriptor(
1483            ctx.clone(),
1484            objects,
1485            pull_into_descriptor,
1486        )?;
1487
1488        // Perform ! ReadableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller).
1489        Self::readable_byte_stream_controller_process_pull_into_descriptors_using_queue(
1490            &ctx, objects,
1491        )
1492    }
1493
1494    pub(super) fn readable_byte_stream_controller_respond_with_new_view<
1495        R: ReadableStreamReader<'js>,
1496    >(
1497        ctx: Ctx<'js>,
1498        mut objects: ReadableByteStreamObjects<'js, R>,
1499        view: ViewBytes<'js>,
1500    ) -> Result<()> {
1501        // Let firstDescriptor be controller.[[pendingPullIntos]][0].
1502        let first_descriptor_index = 0;
1503
1504        let (buffer, byte_length, byte_offset) = view.get_array_buffer()?;
1505
1506        // Let state be controller.[[stream]].[[state]].
1507        match objects.stream.state {
1508            // If state is "closed",
1509            ReadableStreamState::Closed => {
1510                // If view.[[ByteLength]] is not 0, throw a TypeError exception.
1511                if byte_length != 0 {
1512                    return Err(Exception::throw_type(&ctx, "The view's length must be 0 when calling respondWithNewView() on a closed stream"));
1513                }
1514            },
1515            // Otherwise
1516            _ => {
1517                // If view.[[ByteLength]] is 0, throw a TypeError exception.
1518                if byte_length == 0 {
1519                    return Err(Exception::throw_type(&ctx, "The view's length must be greater than 0 when calling respondWithNewView() on a readable stream"));
1520                }
1521            },
1522        };
1523
1524        {
1525            let first_descriptor =
1526                &mut objects.controller.pending_pull_intos[first_descriptor_index];
1527
1528            // If firstDescriptor’s byte offset + firstDescriptor’ bytes filled is not view.[[ByteOffset]], throw a RangeError exception.
1529            if first_descriptor.byte_offset + first_descriptor.bytes_filled != byte_offset {
1530                return Err(Exception::throw_range(
1531                    &ctx,
1532                    "The region specified by view does not match byobRequest",
1533                ));
1534            };
1535
1536            // If firstDescriptor’s buffer byte length is not view.[[ViewedArrayBuffer]].[[ByteLength]], throw a RangeError exception.
1537            if first_descriptor.buffer_byte_length != buffer.len() {
1538                return Err(Exception::throw_range(
1539                    &ctx,
1540                    "The buffer of view has different capacity than byobRequest",
1541                ));
1542            };
1543
1544            // If firstDescriptor’s bytes filled + view.[[ByteLength]] > firstDescriptor’s byte length, throw a RangeError exception.
1545            if first_descriptor.bytes_filled + byte_length > first_descriptor.byte_length {
1546                return Err(Exception::throw_range(
1547                    &ctx,
1548                    "The region specified by view is larger than byobRequest",
1549                ));
1550            }
1551
1552            // Set firstDescriptor’s buffer to ? TransferArrayBuffer(view.[[ViewedArrayBuffer]]).
1553            first_descriptor.buffer = transfer_array_buffer(buffer)?;
1554        }
1555
1556        // Perform ? ReadableByteStreamControllerRespondInternal(controller, viewByteLength).
1557        Self::readable_byte_stream_controller_respond_internal(ctx, objects, byte_length)
1558    }
1559
1560    fn readable_byte_stream_controller_fill_head_pull_into_descriptor<'a>(
1561        &mut self,
1562        size: usize,
1563        pull_into_descriptor_ref: &mut PullIntoDescriptorRefMut<'js, 'a>,
1564    ) {
1565        let pull_into_descriptor = match pull_into_descriptor_ref {
1566            PullIntoDescriptorRefMut::Index(i) => &mut self.pending_pull_intos[*i],
1567            PullIntoDescriptorRefMut::Owned(r) => *r,
1568        };
1569
1570        // Set pullIntoDescriptor’s bytes filled to bytes filled + size.
1571        pull_into_descriptor.bytes_filled += size;
1572    }
1573
1574    fn start_algorithm<R: ReadableStreamReader<'js>>(
1575        ctx: Ctx<'js>,
1576        objects: ReadableByteStreamObjects<'js, R>,
1577        start_algorithm: StartAlgorithm<'js>,
1578    ) -> Result<(
1579        Value<'js>,
1580        ReadableStreamClassObjects<'js, OwnedBorrowMut<'js, Self>, R>,
1581    )> {
1582        let objects_class = objects.into_inner();
1583
1584        Ok((
1585            start_algorithm.call(
1586                ctx,
1587                ReadableStreamControllerClass::ReadableStreamByteController(
1588                    objects_class.controller.clone(),
1589                ),
1590            )?,
1591            objects_class,
1592        ))
1593    }
1594
1595    fn pull_algorithm<R: ReadableStreamReader<'js>>(
1596        ctx: Ctx<'js>,
1597        objects: ReadableByteStreamObjects<'js, R>,
1598    ) -> Result<(
1599        Promise<'js>,
1600        ReadableStreamClassObjects<'js, OwnedBorrowMut<'js, Self>, R>,
1601    )> {
1602        let pull_algorithm = objects
1603            .controller
1604            .pull_algorithm
1605            .clone()
1606            .expect("pull algorithm used after ReadableStreamDefaultControllerClearAlgorithms");
1607        let promise_primordials = objects.stream.promise_primordials.clone();
1608        let objects_class = objects.into_inner();
1609
1610        Ok((
1611            pull_algorithm.call(
1612                ctx,
1613                &promise_primordials,
1614                ReadableStreamControllerClass::ReadableStreamByteController(
1615                    objects_class.controller.clone(),
1616                ),
1617            )?,
1618            objects_class,
1619        ))
1620    }
1621
1622    fn cancel_algorithm<R: ReadableStreamReader<'js>>(
1623        ctx: Ctx<'js>,
1624        objects: ReadableByteStreamObjects<'js, R>,
1625        reason: Value<'js>,
1626    ) -> Result<(
1627        Promise<'js>,
1628        ReadableStreamClassObjects<'js, OwnedBorrowMut<'js, Self>, R>,
1629    )> {
1630        let cancel_algorithm =
1631            objects.controller.cancel_algorithm.clone().expect(
1632                "cancel algorithm used after ReadableStreamDefaultControllerClearAlgorithms",
1633            );
1634        let promise_primordials = objects.stream.promise_primordials.clone();
1635        let objects_class = objects.into_inner();
1636
1637        Ok((
1638            cancel_algorithm.call(ctx, &promise_primordials, reason)?,
1639            objects_class,
1640        ))
1641    }
1642}
1643
1644#[methods(rename_all = "camelCase")]
1645impl<'js> ReadableByteStreamController<'js> {
1646    #[qjs(constructor)]
1647    fn new(ctx: Ctx<'js>) -> Result<Class<'js, Self>> {
1648        Err(Exception::throw_type(&ctx, "Illegal constructor"))
1649    }
1650
1651    // readonly attribute ReadableStreamBYOBRequest? byobRequest;
1652    #[qjs(get, rename = "byobRequest")]
1653    fn byob_request_getter(
1654        ctx: Ctx<'js>,
1655        controller: This<Class<'js, Self>>,
1656    ) -> Result<Null<Class<'js, ReadableStreamBYOBRequest<'js>>>> {
1657        // Use `try_borrow_mut` so that reentrant access during enqueue
1658        // (e.g. via a patched `Object.prototype.then` getter, WPT
1659        // `readable-byte-streams/patched-global`) doesn't hard-error with
1660        // "can't borrow" when the outer enqueue already holds the mut
1661        // borrow. If the controller IS currently borrowed, we can still
1662        // answer correctly by reading the state via an immutable try_borrow;
1663        // materialization is only needed when state is consistent.
1664        if let Ok(owned) = rquickjs::class::OwnedBorrowMut::try_from_class(controller.0.clone()) {
1665            let (request, _) = Self::readable_byte_stream_controller_get_byob_request(ctx, owned)?;
1666            return Ok(request);
1667        }
1668        // Reentrant access mid-enqueue: can't acquire immutable borrow
1669        // either (because enqueue holds mut). Return null — the spec's
1670        // observable state during this transient window is that the
1671        // byob request has been invalidated (the enqueue path clears it
1672        // as pull-into descriptors are filled).
1673        Ok(Null(None))
1674    }
1675
1676    // readonly attribute unrestricted double? desiredSize;
1677    #[qjs(get)]
1678    fn desired_size(&self) -> Null<f64> {
1679        let stream = OwnedBorrow::from_class(self.stream.clone());
1680        self.readable_byte_stream_controller_get_desired_size(&stream)
1681    }
1682
1683    // undefined close();
1684    fn close(ctx: Ctx<'js>, controller: This<OwnedBorrowMut<'js, Self>>) -> Result<()> {
1685        // If this.[[closeRequested]] is true, throw a TypeError exception.
1686        if controller.close_requested {
1687            return Err(Exception::throw_type(&ctx, "close() called more than once"));
1688        }
1689
1690        let objects = ReadableStreamObjects::from_byte_controller(controller.0).refresh_reader();
1691
1692        if !matches!(objects.stream.state, ReadableStreamState::Readable) {
1693            return Err(Exception::throw_type(
1694                &ctx,
1695                "close() called when stream is not readable",
1696            ));
1697        };
1698
1699        // Perform ? ReadableByteStreamControllerClose(this).
1700        Self::readable_byte_stream_controller_close(ctx, objects)?;
1701        Ok(())
1702    }
1703
1704    // undefined enqueue(ArrayBufferView chunk);
1705    fn enqueue(
1706        this: This<OwnedBorrowMut<'js, Self>>,
1707        ctx: Ctx<'js>,
1708        chunk: Value<'js>,
1709    ) -> Result<()> {
1710        let chunk = ViewBytes::from_value(&ctx, &this.function_array_buffer_is_view, Some(&chunk))?;
1711
1712        let (array_buffer, byte_length, _) = chunk.get_array_buffer()?;
1713
1714        // If chunk.[[ByteLength]] is 0, throw a TypeError exception.
1715        if byte_length == 0 {
1716            return Err(Exception::throw_type(
1717                &ctx,
1718                "chunk must have non-zero byteLength",
1719            ));
1720        }
1721
1722        // If chunk.[[ViewedArrayBuffer]].[[ArrayBufferByteLength]] is 0, throw a TypeError exception.
1723        if array_buffer.is_empty() {
1724            return Err(Exception::throw_type(
1725                &ctx,
1726                "chunk must have non-zero buffer byteLength",
1727            ));
1728        }
1729
1730        // If this.[[closeRequested]] is true, throw a TypeError exception.
1731        if this.close_requested {
1732            return Err(Exception::throw_type(&ctx, "stream is closed or draining"));
1733        }
1734
1735        let objects = ReadableStreamObjects::from_byte_controller(this.0).refresh_reader();
1736
1737        // If this.[[stream]].[[state]] is not "readable", throw a TypeError exception.
1738        if !matches!(objects.stream.state, ReadableStreamState::Readable) {
1739            return Err(Exception::throw_type(
1740                &ctx,
1741                "The stream is not in the readable state and cannot be enqueued to",
1742            ));
1743        };
1744
1745        // Return ? ReadableByteStreamControllerEnqueue(this, chunk).
1746        Self::readable_byte_stream_controller_enqueue(&ctx, objects, chunk)?;
1747        Ok(())
1748    }
1749
1750    // undefined error(optional any e);
1751    fn error(
1752        ctx: Ctx<'js>,
1753        controller: This<OwnedBorrowMut<'js, Self>>,
1754        e: Opt<Value<'js>>,
1755    ) -> Result<()> {
1756        let objects = ReadableStreamObjects::from_byte_controller(controller.0).refresh_reader();
1757
1758        // Perform ! ReadableByteStreamControllerError(this, e).
1759        Self::readable_byte_stream_controller_error(objects, e.0.unwrap_or_undefined(&ctx))?;
1760        Ok(())
1761    }
1762}
1763
1764impl<'js> ReadableStreamController<'js> for ReadableByteStreamControllerOwned<'js> {
1765    type Class = ReadableByteStreamControllerClass<'js>;
1766
1767    fn with_controller<C, O>(
1768        self,
1769        ctx: C,
1770        _: impl FnOnce(
1771            C,
1772            ReadableStreamDefaultControllerOwned<'js>,
1773        ) -> Result<(O, ReadableStreamDefaultControllerOwned<'js>)>,
1774        byte: impl FnOnce(
1775            C,
1776            ReadableByteStreamControllerOwned<'js>,
1777        ) -> Result<(O, ReadableByteStreamControllerOwned<'js>)>,
1778    ) -> Result<(O, Self)> {
1779        let (ctx, reader) = byte(ctx, self)?;
1780        Ok((ctx, reader))
1781    }
1782
1783    fn into_inner(self) -> Self::Class {
1784        OwnedBorrowMut::into_inner(self)
1785    }
1786
1787    fn from_class(class: Self::Class) -> Self {
1788        OwnedBorrowMut::from_class(class)
1789    }
1790
1791    fn into_erased(self) -> ReadableStreamControllerOwned<'js> {
1792        ReadableStreamControllerOwned::ReadableStreamByteController(self)
1793    }
1794
1795    fn try_from_erased(erased: ReadableStreamControllerOwned<'js>) -> Option<Self> {
1796        match erased {
1797            ReadableStreamControllerOwned::ReadableStreamDefaultController(_) => None,
1798            ReadableStreamControllerOwned::ReadableStreamByteController(r) => Some(r),
1799        }
1800    }
1801
1802    fn pull_steps(
1803        ctx: &Ctx<'js>,
1804        mut objects: ReadableStreamDefaultReaderObjects<'js, Self>,
1805        read_request: impl ReadableStreamReadRequest<'js> + 'js,
1806    ) -> Result<ReadableStreamDefaultReaderObjects<'js, Self>> {
1807        // If this.[[queueTotalSize]] > 0,
1808        if objects.controller.queue_total_size > 0 {
1809            // Perform ! ReadableByteStreamControllerFillReadRequestFromQueue(this, readRequest).
1810            // Return.
1811            return ReadableByteStreamController::readable_byte_stream_controller_fill_read_request_from_queue(
1812                ctx,
1813                objects,
1814                read_request,
1815            );
1816        }
1817
1818        // Let autoAllocateChunkSize be this.[[autoAllocateChunkSize]].
1819        let auto_allocate_chunk_size = objects.controller.auto_allocate_chunk_size;
1820
1821        // If autoAllocateChunkSize is not undefined,
1822        if let Some(auto_allocate_chunk_size) = auto_allocate_chunk_size {
1823            // Let buffer be Construct(%ArrayBuffer%, « autoAllocateChunkSize »).
1824            let buffer: ArrayBuffer = match objects
1825                .controller
1826                .constructor_array_buffer
1827                .construct((auto_allocate_chunk_size,))
1828            {
1829                // If buffer is an abrupt completion,
1830                Err(Error::Exception) => {
1831                    // Perform readRequest’s error steps, given buffer.[[Value]].
1832                    return read_request.error_steps_typed(objects, ctx.catch());
1833                },
1834                Err(err) => return Err(err),
1835                Ok(buffer) => buffer,
1836            };
1837
1838            // Let pullIntoDescriptor be a new pull-into descriptor with...
1839            let pull_into_descriptor = PullIntoDescriptor {
1840                buffer,
1841                buffer_byte_length: auto_allocate_chunk_size,
1842                byte_offset: 0,
1843                byte_length: auto_allocate_chunk_size,
1844                bytes_filled: 0,
1845                minimum_fill: 1,
1846                element_size: 1,
1847                view_constructor: objects
1848                    .controller
1849                    .array_constructor_primordials
1850                    .constructor_uint8array
1851                    .clone(),
1852                reader_type: PullIntoDescriptorReaderType::Default,
1853            };
1854
1855            // Append pullIntoDescriptor to this.[[pendingPullIntos]].
1856            objects
1857                .controller
1858                .pending_pull_intos
1859                .push_back(pull_into_descriptor);
1860        }
1861
1862        // Perform ! ReadableStreamAddReadRequest(stream, readRequest).
1863        objects
1864            .stream
1865            .readable_stream_add_read_request(&mut objects.reader, read_request);
1866
1867        // Perform ! ReadableByteStreamControllerCallPullIfNeeded(this).
1868        ReadableByteStreamController::readable_byte_stream_controller_call_pull_if_needed(
1869            ctx.clone(),
1870            objects,
1871        )
1872    }
1873
1874    fn cancel_steps<R: ReadableStreamReader<'js>>(
1875        ctx: &Ctx<'js>,
1876        mut objects: ReadableStreamObjects<'js, Self, R>,
1877        reason: Value<'js>,
1878    ) -> Result<(Promise<'js>, ReadableStreamObjects<'js, Self, R>)> {
1879        // Perform ! ReadableByteStreamControllerClearPendingPullIntos(this).
1880        objects
1881            .controller
1882            .readable_byte_stream_controller_clear_pending_pull_intos();
1883
1884        // Perform ! ResetQueue(this).
1885        objects.controller.reset_queue();
1886
1887        // Let result be the result of performing this.[[cancelAlgorithm]], passing in reason.
1888        let (result, objects_class) =
1889            ReadableByteStreamController::cancel_algorithm(ctx.clone(), objects, reason)?;
1890
1891        objects = ReadableStreamObjects::from_class(objects_class);
1892
1893        // Perform ! ReadableByteStreamControllerClearAlgorithms(this).
1894        objects
1895            .controller
1896            .readable_byte_stream_controller_clear_algorithms();
1897
1898        // Return result.
1899        Ok((result, objects))
1900    }
1901
1902    fn release_steps(&mut self) {
1903        // If this.[[pendingPullIntos]] is not empty,
1904        if !self.pending_pull_intos.is_empty() {
1905            // Let firstPendingPullInto be this.[[pendingPullIntos]][0].
1906            let first_pending_pull_into = &mut self.pending_pull_intos[0];
1907
1908            // Set firstPendingPullInto’s reader type to "none".
1909            first_pending_pull_into.reader_type = PullIntoDescriptorReaderType::None;
1910
1911            // Set this.[[pendingPullIntos]] to the list « firstPendingPullInto ».
1912            _ = self.pending_pull_intos.split_off(1);
1913        }
1914    }
1915}
1916
1917#[derive(JsLifetime, Trace, Clone)]
1918#[rquickjs::class]
1919pub(crate) struct ReadableStreamBYOBRequest<'js> {
1920    pub(super) view: Option<ViewBytes<'js>>,
1921    controller: Option<ReadableByteStreamControllerClass<'js>>,
1922}
1923
1924#[methods(rename_all = "camelCase")]
1925impl<'js> ReadableStreamBYOBRequest<'js> {
1926    #[qjs(constructor)]
1927    fn new(ctx: Ctx<'js>) -> Result<Class<'js, Self>> {
1928        Err(Exception::throw_type(&ctx, "Illegal constructor"))
1929    }
1930
1931    #[qjs(get)]
1932    fn view(&self) -> Null<ViewBytes<'js>> {
1933        Null(self.view.clone())
1934    }
1935
1936    fn respond(
1937        ctx: Ctx<'js>,
1938        byob_request: This<OwnedBorrowMut<'js, Self>>,
1939        bytes_written: usize,
1940    ) -> Result<()> {
1941        // If this.[[controller]] is undefined, throw a TypeError exception.
1942        let (controller, view) = match (&byob_request.controller, &byob_request.view) {
1943            (Some(controller), Some(view)) => (controller.clone(), view),
1944            _ => {
1945                return Err(Exception::throw_type(
1946                    &ctx,
1947                    "This BYOB request has been invalidated",
1948                ));
1949            },
1950        };
1951        let (buffer, _, _) = view.get_array_buffer()?;
1952        drop(byob_request);
1953
1954        // If ! IsDetachedBuffer(this.[[view]].[[ArrayBuffer]]) is true, throw a TypeError exception.
1955        // SAFETY: a detachment probe; the slice is never named.
1956        if unsafe { buffer.as_bytes() }.is_none() {
1957            return Err(Exception::throw_type(
1958                &ctx,
1959                "The BYOB request's buffer has been detached and so cannot be used as a response",
1960            ));
1961        }
1962
1963        let objects =
1964            ReadableStreamObjects::from_byte_controller(OwnedBorrowMut::from_class(controller))
1965                .refresh_reader();
1966
1967        // Perform ? ReadableByteStreamControllerRespond(this.[[controller]], bytesWritten).
1968        ReadableByteStreamController::readable_byte_stream_controller_respond(
1969            ctx,
1970            objects,
1971            bytes_written,
1972        )
1973    }
1974
1975    fn respond_with_new_view(
1976        ctx: Ctx<'js>,
1977        byob_request: This<OwnedBorrowMut<'js, Self>>,
1978        view: Opt<Value<'js>>,
1979    ) -> Result<()> {
1980        // If this.[[controller]] is undefined, throw a TypeError exception.
1981        let controller = match &byob_request.controller {
1982            Some(controller) => controller.clone(),
1983            _ => {
1984                return Err(Exception::throw_type(
1985                    &ctx,
1986                    "This BYOB request has been invalidated",
1987                ));
1988            },
1989        };
1990        drop(byob_request);
1991
1992        let controller = OwnedBorrowMut::from_class(controller);
1993
1994        let view = ViewBytes::from_value(
1995            &ctx,
1996            &controller.function_array_buffer_is_view,
1997            view.0.as_ref(),
1998        )?;
1999
2000        let (buffer, _, _) = view.get_array_buffer()?;
2001
2002        // If ! IsDetachedBuffer(view.[[ViewedArrayBuffer]]) is true, throw a TypeError exception.
2003        // SAFETY: a detachment probe; the slice is never named.
2004        if unsafe { buffer.as_bytes() }.is_none() {
2005            return Err(Exception::throw_type(
2006                &ctx,
2007                "The given view's buffer has been detached and so cannot be used as a response",
2008            ));
2009        }
2010
2011        let objects = ReadableStreamObjects::from_byte_controller(controller).refresh_reader();
2012
2013        // Return ? ReadableByteStreamControllerRespondWithNewView(this.[[controller]], view).
2014        ReadableByteStreamController::readable_byte_stream_controller_respond_with_new_view(
2015            ctx, objects, view,
2016        )
2017    }
2018}
2019
2020#[derive(JsLifetime)]
2021pub(super) struct PullIntoDescriptor<'js> {
2022    buffer: ArrayBuffer<'js>,
2023    buffer_byte_length: usize,
2024    byte_offset: usize,
2025    byte_length: usize,
2026    bytes_filled: usize,
2027    minimum_fill: usize,
2028    element_size: usize,
2029    view_constructor: Constructor<'js>,
2030    reader_type: PullIntoDescriptorReaderType,
2031}
2032
2033impl<'js> Trace<'js> for PullIntoDescriptor<'js> {
2034    fn trace<'a>(&self, tracer: Tracer<'a, 'js>) {
2035        self.buffer.trace(tracer);
2036        self.buffer_byte_length.trace(tracer);
2037        self.byte_offset.trace(tracer);
2038        self.byte_length.trace(tracer);
2039        self.bytes_filled.trace(tracer);
2040        self.minimum_fill.trace(tracer);
2041        self.element_size.trace(tracer);
2042        self.view_constructor.trace(tracer);
2043        self.reader_type.trace(tracer);
2044    }
2045}
2046
2047enum PullIntoDescriptorRefMut<'js, 'a> {
2048    Index(usize),
2049    Owned(&'a mut PullIntoDescriptor<'js>),
2050}
2051
2052#[derive(Trace, Clone, Copy)]
2053enum PullIntoDescriptorReaderType {
2054    Default,
2055    Byob,
2056    None,
2057}
2058
2059#[derive(JsLifetime)]
2060struct ReadableByteStreamQueueEntry<'js> {
2061    buffer: ArrayBuffer<'js>,
2062    byte_offset: usize,
2063    byte_length: usize,
2064}
2065
2066impl<'js> Trace<'js> for ReadableByteStreamQueueEntry<'js> {
2067    fn trace<'a>(&self, tracer: Tracer<'a, 'js>) {
2068        self.buffer.trace(tracer);
2069        self.byte_offset.trace(tracer);
2070        self.byte_length.trace(tracer)
2071    }
2072}
2073
2074fn transfer_array_buffer(buffer: ArrayBuffer<'_>) -> Result<ArrayBuffer<'_>> {
2075    buffer.get::<_, Function>("transfer")?.call((This(buffer),))
2076}
2077
2078fn copy_data_block_bytes(
2079    ctx: &Ctx<'_>,
2080    to_block: &ArrayBuffer,
2081    to_index: usize,
2082    from_block: &ArrayBuffer,
2083    from_index: usize,
2084    count: usize,
2085) -> Result<()> {
2086    let to_raw = to_block
2087        .as_raw()
2088        .ok_or(ERROR_MSG_ARRAY_BUFFER_DETACHED)
2089        .or_throw(ctx)?;
2090    let to_slice = unsafe { std::slice::from_raw_parts_mut(to_raw.cast::<u8>().as_ptr(), to_raw.len()) };
2091    let from_raw = from_block
2092        .as_raw()
2093        .ok_or(ERROR_MSG_ARRAY_BUFFER_DETACHED)
2094        .or_throw(ctx)?;
2095    let from_slice = unsafe { std::slice::from_raw_parts(from_raw.cast::<u8>().as_ptr(), from_raw.len()) };
2096
2097    to_slice[to_index..to_index + count]
2098        .copy_from_slice(&from_slice[from_index..from_index + count]);
2099    Ok(())
2100}
2101
2102/// Public API for enqueuing a `Uint8Array` (built from the caller-supplied
2103/// `ArrayBuffer`) into a byte stream controller from Rust code. Used by
2104/// byte-source streams created via `ReadableStream::from_byte_pull_algorithm`.
2105pub fn readable_byte_stream_controller_enqueue_bytes<'js>(
2106    ctx: Ctx<'js>,
2107    controller: ReadableByteStreamControllerClass<'js>,
2108    buffer: ArrayBuffer<'js>,
2109) -> Result<()> {
2110    readable_byte_stream_controller_enqueue_bytes_inner(ctx, controller, buffer, false)
2111}
2112
2113/// Zero-copy variant of [`readable_byte_stream_controller_enqueue_bytes`]:
2114/// the incoming `ArrayBuffer` is NOT transferred/detached before being
2115/// queued. The producer keeps the buffer alive through the stream's
2116/// `'js` queue entry, so consumers get a `Uint8Array` that views directly
2117/// into the producer's storage.
2118///
2119/// Only call this when the caller can guarantee the backing allocation
2120/// won't be mutated out from under readers (e.g. `Blob.stream()`, where
2121/// the blob's `ArrayBuffer` is never written after construction). For the
2122/// normal spec-compliant flow that detaches the source, use
2123/// [`readable_byte_stream_controller_enqueue_bytes`].
2124pub fn readable_byte_stream_controller_enqueue_bytes_borrowed<'js>(
2125    ctx: Ctx<'js>,
2126    controller: ReadableByteStreamControllerClass<'js>,
2127    buffer: ArrayBuffer<'js>,
2128) -> Result<()> {
2129    readable_byte_stream_controller_enqueue_bytes_inner(ctx, controller, buffer, true)
2130}
2131
2132fn readable_byte_stream_controller_enqueue_bytes_inner<'js>(
2133    ctx: Ctx<'js>,
2134    controller: ReadableByteStreamControllerClass<'js>,
2135    buffer: ArrayBuffer<'js>,
2136    skip_transfer: bool,
2137) -> Result<()> {
2138    let byte_length = buffer.len();
2139    if byte_length == 0 {
2140        return Ok(());
2141    }
2142    let view = rquickjs::TypedArray::<u8>::from_arraybuffer(buffer)?;
2143    let borrow = OwnedBorrowMut::from_class(controller);
2144    let chunk = ViewBytes::from_value(
2145        &ctx,
2146        &borrow.function_array_buffer_is_view,
2147        Some(&view.into_value()),
2148    )?;
2149    let objects = ReadableStreamObjects::from_byte_controller(borrow).refresh_reader();
2150    if skip_transfer {
2151        ReadableByteStreamController::readable_byte_stream_controller_enqueue_borrowed(
2152            &ctx, objects, chunk,
2153        )?;
2154    } else {
2155        ReadableByteStreamController::readable_byte_stream_controller_enqueue(
2156            &ctx, objects, chunk,
2157        )?;
2158    }
2159    Ok(())
2160}
2161
2162/// Public API for closing a byte stream controller from Rust code.
2163pub fn readable_byte_stream_controller_close_stream<'js>(
2164    ctx: Ctx<'js>,
2165    controller: ReadableByteStreamControllerClass<'js>,
2166) -> Result<()> {
2167    let borrow = OwnedBorrowMut::from_class(controller);
2168    let objects = ReadableStreamObjects::from_byte_controller(borrow).refresh_reader();
2169    ReadableByteStreamController::readable_byte_stream_controller_close(ctx, objects)?;
2170    Ok(())
2171}