Skip to main content

script/dom/stream/
readablestreamdefaultreader.rs

1/* This Source Code Form is subject to the terms of the Mozilla Public
2 * License, v. 2.0. If a copy of the MPL was not distributed with this
3 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
4
5use std::cell::Cell;
6use std::collections::VecDeque;
7use std::mem;
8use std::rc::Rc;
9
10use dom_struct::dom_struct;
11use js::context::JSContext;
12use js::conversions::{ConversionResult, FromJSValConvertible};
13use js::jsapi::Heap;
14use js::jsval::{JSVal, UndefinedValue};
15use js::realm::CurrentRealm;
16use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
17use js::typedarray::Uint8Array;
18use script_bindings::cell::DomRefCell;
19use script_bindings::reflector::{Reflector, reflect_dom_object, reflect_dom_object_with_proto};
20
21use super::byteteereadrequest::ByteTeeReadRequest;
22use super::readablebytestreamcontroller::ReadableByteStreamController;
23use crate::dom::bindings::codegen::Bindings::ReadableStreamDefaultReaderBinding::{
24    ReadableStreamDefaultReaderMethods, ReadableStreamReadResult,
25};
26use crate::dom::bindings::conversions::ConversionBehavior;
27use crate::dom::bindings::error::{Error, ErrorToJsval, Fallible};
28use crate::dom::bindings::reflector::DomGlobal;
29use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
30use crate::dom::bindings::trace::RootedTraceableBox;
31use crate::dom::globalscope::GlobalScope;
32use crate::dom::promise::{Promise, RootedPromise, TracedPromise};
33use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
34use crate::dom::readablestream::ReadableStream;
35use crate::dom::stream::defaultteereadrequest::DefaultTeeReadRequest;
36use crate::dom::stream::readablestreamgenericreader::ReadableStreamGenericReader;
37use crate::dom::types::ReadableStreamDefaultController;
38use crate::realms::enter_auto_realm;
39
40type ReadAllBytesSuccessSteps = dyn Fn(&mut js::context::JSContext, &[u8]);
41type ReadAllBytesFailureSteps = dyn Fn(&mut js::context::JSContext, SafeHandleValue);
42
43impl js::gc::Rootable for ContinueReadMicrotask {}
44
45/// Microtask handler to continue the read loop without recursion.
46/// Spec note: "This recursion could potentially cause a stack overflow
47/// if implemented directly. Implementations will need to mitigate this,
48/// e.g. by using a non-recursive variant of this algorithm, or queuing
49/// a microtask…"
50#[derive(Clone, JSTraceable, MallocSizeOf)]
51#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
52struct ContinueReadMicrotask {
53    reader: Dom<ReadableStreamDefaultReader>,
54    request: ReadRequest,
55}
56
57impl Callback for ContinueReadMicrotask {
58    fn callback(&self, cx: &mut CurrentRealm, _v: SafeHandleValue) {
59        // https://streams.spec.whatwg.org/#ref-for-read-loop%E2%91%A0
60        // Note: continuing the read-loop from inside a micro-task to break recursion.
61        self.reader.read(cx, &self.request);
62    }
63}
64
65/// <https://streams.spec.whatwg.org/#read-loop>
66fn read_loop(
67    cx: &mut js::context::JSContext,
68    reader: &ReadableStreamDefaultReader,
69    success_steps: Rc<ReadAllBytesSuccessSteps>,
70    failure_steps: Rc<ReadAllBytesFailureSteps>,
71) {
72    // For the purposes of the above algorithm, to read-loop given reader,
73    // bytes, successSteps, and failureSteps:
74
75    // Step 1 .Let readRequest be a new read request with the following items:
76    rooted!(&in(cx) let req = ReadRequest::ReadLoop {
77        success_steps,
78        failure_steps,
79        reader: Dom::from_ref(reader),
80        bytes: Rc::new(DomRefCell::new(Vec::new())),
81    });
82    // Step 2 .Perform ! ReadableStreamDefaultReaderRead(reader, readRequest).
83    reader.read(cx, &req);
84}
85
86/// <https://streams.spec.whatwg.org/#read-request>
87#[derive(Clone, JSTraceable, MallocSizeOf)]
88#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
89pub(crate) enum ReadRequest {
90    /// <https://streams.spec.whatwg.org/#default-reader-read>
91    Read(TracedPromise),
92    /// <https://streams.spec.whatwg.org/#ref-for-read-request%E2%91%A2>
93    DefaultTee {
94        tee_read_request: Dom<DefaultTeeReadRequest>,
95    },
96    /// Spec read loop variant, driven by read-request steps (no Promise).
97    /// <https://streams.spec.whatwg.org/#read-loop>
98    ReadLoop {
99        #[ignore_malloc_size_of = "dyn Fn"]
100        #[no_trace]
101        success_steps: Rc<ReadAllBytesSuccessSteps>,
102        #[ignore_malloc_size_of = "dyn Fn"]
103        #[no_trace]
104        failure_steps: Rc<ReadAllBytesFailureSteps>,
105        reader: Dom<ReadableStreamDefaultReader>,
106        #[conditional_malloc_size_of]
107        bytes: Rc<DomRefCell<Vec<u8>>>,
108    },
109    ByteTee {
110        byte_tee_read_request: Dom<ByteTeeReadRequest>,
111    },
112}
113
114impl js::rust::Rootable for ReadRequest {}
115
116impl ReadRequest {
117    /// <https://streams.spec.whatwg.org/#read-request-chunk-steps>
118    pub(crate) fn chunk_steps(
119        &self,
120        cx: &mut js::context::JSContext,
121        chunk: RootedTraceableBox<Heap<JSVal>>,
122        global: &GlobalScope,
123    ) {
124        match self {
125            ReadRequest::Read(promise) => {
126                // chunk steps, given chunk
127                // Resolve promise with «[ "value" → chunk, "done" → false ]».
128                promise.resolve_native(
129                    cx,
130                    &ReadableStreamReadResult {
131                        done: Some(false),
132                        value: chunk.into_box(),
133                    },
134                );
135            },
136            ReadRequest::DefaultTee { tee_read_request } => {
137                tee_read_request.enqueue_chunk_steps(cx, chunk);
138            },
139            ReadRequest::ByteTee {
140                byte_tee_read_request,
141            } => {
142                byte_tee_read_request.enqueue_chunk_steps(cx, global, chunk);
143            },
144            ReadRequest::ReadLoop {
145                success_steps: _,
146                failure_steps,
147                reader,
148                bytes,
149            } => {
150                // Spec: chunk steps, given chunk
151                let global = reader.global();
152
153                // Step 1. If chunk is not a Uint8Array object, call failureSteps with a TypeError and abort.
154                auto_root!(&in(cx) let array = match Uint8Array::from_jsval(cx, chunk.handle(), ()) {
155                    Ok(ConversionResult::Success(array)) => array,
156                    _ => {
157                        rooted!(&in(cx) let mut v = UndefinedValue());
158                        Error::Type(c"Chunk is not a Uint8Array.".to_owned())
159                            .to_jsval(cx, &global, v.handle_mut());
160                        (failure_steps)(cx, v.handle());
161                        return;
162                    },
163                });
164
165                // Step 2. Append the bytes represented by chunk to bytes.
166                if array.is_shared() {
167                    let chunk_bytes = match Vec::<u8>::from_jsval(
168                        cx,
169                        chunk.handle(),
170                        ConversionBehavior::EnforceRange,
171                    ) {
172                        Ok(ConversionResult::Success(chunk_bytes)) => chunk_bytes,
173                        _ => {
174                            rooted!(&in(cx) let mut v = UndefinedValue());
175                            Error::Type(c"Unknown format for bytes read.".to_owned()).to_jsval(
176                                cx,
177                                &global,
178                                v.handle_mut(),
179                            );
180                            (failure_steps)(cx, v.handle());
181                            return;
182                        },
183                    };
184                    bytes.borrow_mut().extend_from_slice(&chunk_bytes);
185                } else {
186                    bytes
187                        .borrow_mut()
188                        .extend_from_slice(array.as_slice_safe(cx.no_gc()).unwrap_or(&[]));
189                }
190
191                // Step 3. Read-loop given reader, bytes, successSteps, and failureSteps.
192                // Spec note: Avoid direct recursion; queue into a microtask.
193                // Resolving the promise will queue a microtask to call into the native handler.
194                let tick = Promise::new(cx, &global);
195                tick.resolve_native(cx, &());
196
197                let handler = PromiseNativeHandler::new(
198                    cx,
199                    &global,
200                    Some(Box::new(ContinueReadMicrotask {
201                        reader: Dom::from_ref(reader),
202                        request: self.clone(),
203                    })),
204                    None,
205                );
206
207                let mut realm = enter_auto_realm(cx, &*global);
208                let cx = &mut realm.current_realm();
209                tick.append_native_handler(cx, &handler);
210            },
211        }
212    }
213
214    /// <https://streams.spec.whatwg.org/#read-request-close-steps>
215    pub(crate) fn close_steps(&self, cx: &mut js::context::JSContext) {
216        match self {
217            ReadRequest::Read(promise) => {
218                // close steps
219                // Resolve promise with «[ "value" → undefined, "done" → true ]».
220                promise.resolve_native(
221                    cx,
222                    &ReadableStreamReadResult {
223                        done: Some(true),
224                        value: Heap::boxed(UndefinedValue()),
225                    },
226                );
227            },
228            ReadRequest::DefaultTee { tee_read_request } => {
229                tee_read_request.close_steps(cx);
230            },
231            ReadRequest::ByteTee {
232                byte_tee_read_request,
233            } => {
234                byte_tee_read_request
235                    .close_steps(cx)
236                    .expect("ByteTeeReadRequest close steps should not fail");
237            },
238            ReadRequest::ReadLoop {
239                success_steps,
240                reader,
241                bytes,
242                ..
243            } => {
244                // Step 1. Call successSteps with bytes.
245                (success_steps)(cx, &bytes.borrow());
246
247                reader
248                    .release(cx)
249                    .expect("Releasing the read-all-bytes reader should succeed");
250            },
251        }
252    }
253
254    /// <https://streams.spec.whatwg.org/#read-request-error-steps>
255    pub(crate) fn error_steps(&self, cx: &mut js::context::JSContext, e: SafeHandleValue) {
256        match self {
257            ReadRequest::Read(promise) => {
258                // error steps, given e
259                // Reject promise with e.
260                promise.reject_native(cx, &e)
261            },
262            ReadRequest::DefaultTee { tee_read_request } => {
263                tee_read_request.error_steps();
264            },
265            ReadRequest::ByteTee {
266                byte_tee_read_request,
267            } => {
268                byte_tee_read_request.error_steps();
269            },
270            ReadRequest::ReadLoop {
271                failure_steps,
272                reader,
273                ..
274            } => {
275                // Step 1. Call failureSteps with e.
276                (failure_steps)(cx, e);
277
278                reader
279                    .release(cx)
280                    .expect("Releasing the read-all-bytes reader should succeed");
281            },
282        }
283    }
284}
285
286/// The rejection handler for
287/// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamtee>
288#[derive(Clone, JSTraceable, MallocSizeOf)]
289#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
290struct ByteTeeClosedPromiseRejectionHandler {
291    branch_1_controller: Dom<ReadableByteStreamController>,
292    branch_2_controller: Dom<ReadableByteStreamController>,
293    #[conditional_malloc_size_of]
294    canceled_1: Rc<Cell<bool>>,
295    #[conditional_malloc_size_of]
296    canceled_2: Rc<Cell<bool>>,
297    cancel_promise: TracedPromise,
298    #[conditional_malloc_size_of]
299    reader_version: Rc<Cell<u64>>,
300    expected_version: u64,
301}
302
303impl Callback for ByteTeeClosedPromiseRejectionHandler {
304    /// Continuation of <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamtee>
305    /// Upon rejection of reader.[[closedPromise]] with reason r,
306    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
307        // If thisReader is not the current `reader`, return.
308        if self.reader_version.get() != self.expected_version {
309            return;
310        }
311
312        // Perform ! ReadableByteStreamControllerError(branch1.[[controller]], r).
313        self.branch_1_controller.error(cx, v);
314
315        // Perform ! ReadableByteStreamControllerError(branch2.[[controller]], r).
316        self.branch_2_controller.error(cx, v);
317
318        // If canceled1 is false or canceled2 is false, resolve cancelPromise with undefined.
319        if !self.canceled_1.get() || !self.canceled_2.get() {
320            self.cancel_promise.resolve_native(cx, &());
321        }
322    }
323}
324
325/// The rejection handler for
326/// <https://streams.spec.whatwg.org/#readable-stream-tee>
327#[derive(Clone, JSTraceable, MallocSizeOf)]
328#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
329struct DefaultTeeClosedPromiseRejectionHandler {
330    branch_1_controller: Dom<ReadableStreamDefaultController>,
331    branch_2_controller: Dom<ReadableStreamDefaultController>,
332    #[conditional_malloc_size_of]
333    canceled_1: Rc<Cell<bool>>,
334    #[conditional_malloc_size_of]
335    canceled_2: Rc<Cell<bool>>,
336    cancel_promise: TracedPromise,
337}
338
339impl Callback for DefaultTeeClosedPromiseRejectionHandler {
340    /// Continuation of <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaulttee>
341    /// Upon rejection of reader.[[closedPromise]] with reason r,
342    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
343        // Perform ! ReadableStreamDefaultControllerError(branch_1.[[controller]], r).
344        self.branch_1_controller.error(cx, v);
345        // Perform ! ReadableStreamDefaultControllerError(branch_2.[[controller]], r).
346        self.branch_2_controller.error(cx, v);
347
348        // If canceled_1 is false or canceled_2 is false, resolve cancelPromise with undefined.
349        if !self.canceled_1.get() || !self.canceled_2.get() {
350            self.cancel_promise.resolve_native(cx, &());
351        }
352    }
353}
354
355/// <https://streams.spec.whatwg.org/#readablestreamdefaultreader>
356#[dom_struct]
357pub(crate) struct ReadableStreamDefaultReader {
358    reflector_: Reflector,
359
360    /// <https://streams.spec.whatwg.org/#readablestreamgenericreader-stream>
361    stream: MutNullableDom<ReadableStream>,
362
363    read_requests: DomRefCell<VecDeque<ReadRequest>>,
364
365    /// <https://streams.spec.whatwg.org/#readablestreamgenericreader-closedpromise>
366    closed_promise: DomRefCell<TracedPromise>,
367}
368
369impl ReadableStreamDefaultReader {
370    fn new_with_proto(
371        cx: &mut JSContext,
372        global: &GlobalScope,
373        proto: Option<SafeHandleObject>,
374    ) -> DomRoot<ReadableStreamDefaultReader> {
375        let closed_promise = Promise::new(cx, global);
376        reflect_dom_object_with_proto(
377            cx,
378            Box::new(ReadableStreamDefaultReader::new_inherited(&closed_promise)),
379            global,
380            proto,
381        )
382    }
383
384    fn new_inherited(promise: &RootedPromise) -> ReadableStreamDefaultReader {
385        ReadableStreamDefaultReader {
386            reflector_: Reflector::new(),
387            stream: MutNullableDom::new(None),
388            read_requests: DomRefCell::new(Default::default()),
389            closed_promise: DomRefCell::new(promise.to_traced()),
390        }
391    }
392
393    pub(crate) fn new(
394        cx: &mut JSContext,
395        global: &GlobalScope,
396    ) -> DomRoot<ReadableStreamDefaultReader> {
397        let closed_promise = Promise::new(cx, global);
398        reflect_dom_object(cx, Box::new(Self::new_inherited(&closed_promise)), global)
399    }
400
401    /// <https://streams.spec.whatwg.org/#set-up-readable-stream-default-reader>
402    pub(crate) fn set_up(
403        &self,
404        cx: &mut JSContext,
405        stream: &ReadableStream,
406        global: &GlobalScope,
407    ) -> Fallible<()> {
408        // If ! IsReadableStreamLocked(stream) is true, throw a TypeError exception.
409        if stream.is_locked() {
410            return Err(Error::Type(c"stream is locked".to_owned()));
411        }
412        // Perform ! ReadableStreamReaderGenericInitialize(reader, stream).
413
414        self.generic_initialize(cx, global, stream);
415
416        // Set reader.[[readRequests]] to a new empty list.
417        self.read_requests.borrow_mut().clear();
418
419        Ok(())
420    }
421
422    /// <https://streams.spec.whatwg.org/#readable-stream-close>
423    pub(crate) fn close(&self, cx: &mut js::context::JSContext) {
424        // Resolve reader.[[closedPromise]] with undefined.
425        self.closed_promise.borrow().resolve_native(cx, &());
426        // If reader implements ReadableStreamDefaultReader,
427        // Let readRequests be reader.[[readRequests]].
428        rooted!(&in(cx) let mut read_requests = self.take_read_requests());
429        // Set reader.[[readRequests]] to an empty list.
430        // For each readRequest of readRequests,
431        for request in read_requests.iter() {
432            // Perform readRequest’s close steps.
433            request.close_steps(cx);
434        }
435    }
436
437    /// <https://streams.spec.whatwg.org/#readable-stream-add-read-request>
438    pub(crate) fn add_read_request(&self, read_request: &ReadRequest) {
439        self.read_requests
440            .borrow_mut()
441            .push_back(read_request.clone());
442    }
443
444    /// <https://streams.spec.whatwg.org/#readable-stream-get-num-read-requests>
445    pub(crate) fn get_num_read_requests(&self) -> usize {
446        self.read_requests.borrow().len()
447    }
448
449    /// <https://streams.spec.whatwg.org/#readable-stream-error>
450    pub(crate) fn error(&self, cx: &mut js::context::JSContext, e: SafeHandleValue) {
451        // Reject reader.[[closedPromise]] with e.
452        self.closed_promise.borrow().reject_native(cx, &e);
453
454        // Set reader.[[closedPromise]].[[PromiseIsHandled]] to true.
455        self.closed_promise.borrow().set_promise_is_handled(cx);
456
457        // Perform ! ReadableStreamDefaultReaderErrorReadRequests(reader, e).
458        self.error_read_requests(cx, e);
459    }
460
461    /// The removal steps of <https://streams.spec.whatwg.org/#readable-stream-fulfill-read-request>
462    pub(crate) fn remove_read_request(&self) -> ReadRequest {
463        self.read_requests
464            .borrow_mut()
465            .pop_front()
466            .expect("Reader must have read request when remove is called into.")
467    }
468
469    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaultreaderrelease>
470    pub(crate) fn release(&self, cx: &mut js::context::JSContext) -> Fallible<()> {
471        // Perform ! ReadableStreamReaderGenericRelease(reader).
472        self.generic_release(cx).expect("Generic release failed");
473        // Let e be a new TypeError exception.
474        rooted!(&in(cx) let mut error = UndefinedValue());
475        Error::Type(c"Reader is released".to_owned()).to_jsval(
476            cx,
477            &self.global(),
478            error.handle_mut(),
479        );
480
481        // Perform ! ReadableStreamDefaultReaderErrorReadRequests(reader, e).
482        self.error_read_requests(cx, error.handle());
483        Ok(())
484    }
485
486    fn take_read_requests(&self) -> VecDeque<ReadRequest> {
487        mem::take(&mut *self.read_requests.borrow_mut())
488    }
489
490    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaultreadererrorreadrequests>
491    fn error_read_requests(&self, cx: &mut js::context::JSContext, rval: SafeHandleValue) {
492        // step 1
493        rooted!(&in(cx) let mut read_requests = self.take_read_requests());
494
495        // step 2 & 3
496        for request in read_requests.iter() {
497            request.error_steps(cx, rval);
498        }
499    }
500
501    /// <https://streams.spec.whatwg.org/#readable-stream-default-reader-read>
502    pub(crate) fn read(&self, cx: &mut js::context::JSContext, read_request: &ReadRequest) {
503        // Let stream be reader.[[stream]].
504
505        // Assert: stream is not undefined.
506        assert!(self.stream.get().is_some());
507
508        let stream = self.stream.get().unwrap();
509
510        // Set stream.[[disturbed]] to true.
511        stream.set_is_disturbed(true);
512        // If stream.[[state]] is "closed", perform readRequest’s close steps.
513        if stream.is_closed() {
514            read_request.close_steps(cx);
515        } else if stream.is_errored() {
516            // Otherwise, if stream.[[state]] is "errored",
517            // perform readRequest’s error steps given stream.[[storedError]].
518            rooted!(&in(cx) let mut error = UndefinedValue());
519            stream.get_stored_error(error.handle_mut());
520            read_request.error_steps(cx, error.handle());
521        } else {
522            // Otherwise
523            // Assert: stream.[[state]] is "readable".
524            assert!(stream.is_readable());
525            // Perform ! stream.[[controller]].[[PullSteps]](readRequest).
526            stream.perform_pull_steps(cx, read_request);
527        }
528    }
529
530    /// Attach the byte-tee error handler to this reader's closedPromise.
531    /// Used by ReadableByteStreamTee.
532    #[allow(clippy::too_many_arguments)]
533    pub(crate) fn byte_tee_append_native_handler_to_closed_promise(
534        &self,
535        cx: &mut js::context::JSContext,
536        branch_1: &ReadableStream,
537        branch_2: &ReadableStream,
538        canceled_1: Rc<Cell<bool>>,
539        canceled_2: Rc<Cell<bool>>,
540        cancel_promise: &RootedPromise,
541        reader_version: Rc<Cell<u64>>,
542        expected_version: u64,
543    ) {
544        // Note: for byte tee we always operate on *byte controllers*.
545        let branch_1_controller = branch_1.get_byte_controller();
546        let branch_2_controller = branch_2.get_byte_controller();
547
548        let global = self.global();
549        let handler = PromiseNativeHandler::new(
550            cx,
551            &global,
552            None,
553            Some(Box::new(ByteTeeClosedPromiseRejectionHandler {
554                branch_1_controller: Dom::from_ref(&branch_1_controller),
555                branch_2_controller: Dom::from_ref(&branch_2_controller),
556                canceled_1,
557                canceled_2,
558                cancel_promise: cancel_promise.to_traced(),
559                reader_version,
560                expected_version,
561            })),
562        );
563
564        let mut realm = enter_auto_realm(cx, &*global);
565        let cx = &mut realm.current_realm();
566
567        self.closed_promise
568            .borrow()
569            .append_native_handler(cx, &handler);
570    }
571
572    /// <https://streams.spec.whatwg.org/#ref-for-readablestreamgenericreader-closedpromise%E2%91%A1>
573    pub(crate) fn default_tee_append_native_handler_to_closed_promise(
574        &self,
575        cx: &mut js::context::JSContext,
576        branch_1: &ReadableStream,
577        branch_2: &ReadableStream,
578        canceled_1: Rc<Cell<bool>>,
579        canceled_2: Rc<Cell<bool>>,
580        cancel_promise: &RootedPromise,
581    ) {
582        let branch_1_controller = branch_1.get_default_controller();
583
584        let branch_2_controller = branch_2.get_default_controller();
585
586        let global = self.global();
587        let handler = PromiseNativeHandler::new(
588            cx,
589            &global,
590            None,
591            Some(Box::new(DefaultTeeClosedPromiseRejectionHandler {
592                branch_1_controller: Dom::from_ref(&branch_1_controller),
593                branch_2_controller: Dom::from_ref(&branch_2_controller),
594                canceled_1,
595                canceled_2,
596                cancel_promise: cancel_promise.to_traced(),
597            })),
598        );
599
600        let mut realm = enter_auto_realm(cx, &*global);
601        let cx = &mut realm.current_realm();
602
603        self.closed_promise
604            .borrow()
605            .append_native_handler(cx, &handler);
606    }
607
608    /// <https://streams.spec.whatwg.org/#readablestreamdefaultreader-read-all-bytes>
609    pub(crate) fn read_all_bytes(
610        &self,
611        cx: &mut js::context::JSContext,
612        success_steps: Rc<ReadAllBytesSuccessSteps>,
613        failure_steps: Rc<ReadAllBytesFailureSteps>,
614    ) {
615        // To read all bytes from a ReadableStreamDefaultReader reader,
616        // given successSteps, which is an algorithm accepting a byte sequence,
617        // and failureSteps, which is an algorithm accepting a JavaScript value:
618        // read-loop given reader, a new byte sequence, successSteps, and failureSteps.
619        read_loop(cx, self, success_steps, failure_steps);
620    }
621
622    /// step 3 of <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerprocessreadrequestsusingqueue>
623    pub(crate) fn process_read_requests(
624        &self,
625        cx: &mut js::context::JSContext,
626        controller: &ReadableByteStreamController,
627    ) -> Fallible<()> {
628        // While reader.[[readRequests]] is not empty,
629        while !self.read_requests.borrow().is_empty() {
630            // If controller.[[queueTotalSize]] is 0, return.
631            if controller.get_queue_total_size() == 0.0 {
632                return Ok(());
633            }
634
635            // Let readRequest be reader.[[readRequests]][0].
636            // Remove entry from controller.[[queue]].
637            rooted!(&in(cx) let read_request = self.remove_read_request());
638
639            // Perform ! ReadableByteStreamControllerFillReadRequestFromQueue(controller, readRequest).
640            controller
641                .fill_read_request_from_queue(cx, &read_request)
642                .expect("Fill read request from queue failed");
643        }
644        Ok(())
645    }
646}
647
648impl ReadableStreamDefaultReaderMethods<crate::DomTypeHolder> for ReadableStreamDefaultReader {
649    /// <https://streams.spec.whatwg.org/#default-reader-constructor>
650    fn Constructor(
651        cx: &mut JSContext,
652        global: &GlobalScope,
653        proto: Option<SafeHandleObject>,
654        stream: &ReadableStream,
655    ) -> Fallible<DomRoot<Self>> {
656        let reader = Self::new_with_proto(cx, global, proto);
657
658        // Perform ? SetUpReadableStreamDefaultReader(this, stream).
659        reader.set_up(cx, stream, global)?;
660
661        Ok(reader)
662    }
663
664    /// <https://streams.spec.whatwg.org/#default-reader-read>
665    fn Read(&self, cx: &mut js::context::JSContext) -> RootedPromise {
666        // If this.[[stream]] is undefined, return a promise rejected with a TypeError exception.
667        if self.stream.get().is_none() {
668            rooted!(&in(cx) let mut error = UndefinedValue());
669            Error::Type(c"stream is undefined".to_owned()).to_jsval(
670                cx,
671                &self.global(),
672                error.handle_mut(),
673            );
674            return Promise::new_rejected(cx, &self.global(), error.handle());
675        }
676        // Let promise be a new promise.
677        let promise = Promise::new(cx, &self.global());
678
679        // Let readRequest be a new read request with the following items:
680        // chunk steps, given chunk
681        // Resolve promise with «[ "value" → chunk, "done" → false ]».
682        //
683        // close steps
684        // Resolve promise with «[ "value" → undefined, "done" → true ]».
685        //
686        // error steps, given e
687        // Reject promise with e.
688
689        rooted!(&in(cx) let read_request = ReadRequest::Read(promise.to_traced()));
690
691        // Perform ! ReadableStreamDefaultReaderRead(this, readRequest).
692        self.read(cx, &read_request);
693
694        // Return promise.
695        promise
696    }
697
698    /// <https://streams.spec.whatwg.org/#default-reader-release-lock>
699    fn ReleaseLock(&self, cx: &mut js::context::JSContext) -> Fallible<()> {
700        if self.stream.get().is_none() {
701            // Step 1: If this.[[stream]] is undefined, return.
702            return Ok(());
703        }
704
705        // Step 2: Perform !ReadableStreamDefaultReaderRelease(this).
706        self.release(cx)
707    }
708
709    /// <https://streams.spec.whatwg.org/#generic-reader-closed>
710    fn Closed(&self, cx: &JSContext) -> RootedPromise {
711        self.closed(cx)
712    }
713
714    /// <https://streams.spec.whatwg.org/#generic-reader-cancel>
715    fn Cancel(&self, cx: &mut js::context::JSContext, reason: SafeHandleValue) -> RootedPromise {
716        self.generic_cancel(cx, &self.global(), reason)
717    }
718}
719
720impl ReadableStreamGenericReader for ReadableStreamDefaultReader {
721    fn get_closed_promise(&self, cx: &JSContext) -> RootedPromise {
722        self.closed_promise.borrow().root(cx)
723    }
724
725    fn set_closed_promise(&self, promise: &RootedPromise) {
726        *self.closed_promise.borrow_mut() = promise.to_traced();
727    }
728
729    fn set_stream(&self, stream: Option<&ReadableStream>) {
730        self.stream.set(stream);
731    }
732
733    fn get_stream(&self) -> Option<DomRoot<ReadableStream>> {
734        self.stream.get()
735    }
736
737    fn as_default_reader(&self) -> Option<&ReadableStreamDefaultReader> {
738        Some(self)
739    }
740}