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, Vec<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                let bytes = mem::take(&mut *bytes.borrow_mut());
246                (success_steps)(cx, bytes);
247
248                reader
249                    .release(cx)
250                    .expect("Releasing the read-all-bytes reader should succeed");
251            },
252        }
253    }
254
255    /// <https://streams.spec.whatwg.org/#read-request-error-steps>
256    pub(crate) fn error_steps(&self, cx: &mut js::context::JSContext, e: SafeHandleValue) {
257        match self {
258            ReadRequest::Read(promise) => {
259                // error steps, given e
260                // Reject promise with e.
261                promise.reject_native(cx, &e)
262            },
263            ReadRequest::DefaultTee { tee_read_request } => {
264                tee_read_request.error_steps();
265            },
266            ReadRequest::ByteTee {
267                byte_tee_read_request,
268            } => {
269                byte_tee_read_request.error_steps();
270            },
271            ReadRequest::ReadLoop {
272                failure_steps,
273                reader,
274                ..
275            } => {
276                // Step 1. Call failureSteps with e.
277                (failure_steps)(cx, e);
278
279                reader
280                    .release(cx)
281                    .expect("Releasing the read-all-bytes reader should succeed");
282            },
283        }
284    }
285}
286
287/// The rejection handler for
288/// <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamtee>
289#[derive(Clone, JSTraceable, MallocSizeOf)]
290#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
291struct ByteTeeClosedPromiseRejectionHandler {
292    branch_1_controller: Dom<ReadableByteStreamController>,
293    branch_2_controller: Dom<ReadableByteStreamController>,
294    #[conditional_malloc_size_of]
295    canceled_1: Rc<Cell<bool>>,
296    #[conditional_malloc_size_of]
297    canceled_2: Rc<Cell<bool>>,
298    cancel_promise: TracedPromise,
299    #[conditional_malloc_size_of]
300    reader_version: Rc<Cell<u64>>,
301    expected_version: u64,
302}
303
304impl Callback for ByteTeeClosedPromiseRejectionHandler {
305    /// Continuation of <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamtee>
306    /// Upon rejection of reader.[[closedPromise]] with reason r,
307    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
308        // If thisReader is not the current `reader`, return.
309        if self.reader_version.get() != self.expected_version {
310            return;
311        }
312
313        // Perform ! ReadableByteStreamControllerError(branch1.[[controller]], r).
314        self.branch_1_controller.error(cx, v);
315
316        // Perform ! ReadableByteStreamControllerError(branch2.[[controller]], r).
317        self.branch_2_controller.error(cx, v);
318
319        // If canceled1 is false or canceled2 is false, resolve cancelPromise with undefined.
320        if !self.canceled_1.get() || !self.canceled_2.get() {
321            self.cancel_promise.resolve_native(cx, &());
322        }
323    }
324}
325
326/// The rejection handler for
327/// <https://streams.spec.whatwg.org/#readable-stream-tee>
328#[derive(Clone, JSTraceable, MallocSizeOf)]
329#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
330struct DefaultTeeClosedPromiseRejectionHandler {
331    branch_1_controller: Dom<ReadableStreamDefaultController>,
332    branch_2_controller: Dom<ReadableStreamDefaultController>,
333    #[conditional_malloc_size_of]
334    canceled_1: Rc<Cell<bool>>,
335    #[conditional_malloc_size_of]
336    canceled_2: Rc<Cell<bool>>,
337    cancel_promise: TracedPromise,
338}
339
340impl Callback for DefaultTeeClosedPromiseRejectionHandler {
341    /// Continuation of <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaulttee>
342    /// Upon rejection of reader.[[closedPromise]] with reason r,
343    fn callback(&self, cx: &mut CurrentRealm, v: SafeHandleValue) {
344        // Perform ! ReadableStreamDefaultControllerError(branch_1.[[controller]], r).
345        self.branch_1_controller.error(cx, v);
346        // Perform ! ReadableStreamDefaultControllerError(branch_2.[[controller]], r).
347        self.branch_2_controller.error(cx, v);
348
349        // If canceled_1 is false or canceled_2 is false, resolve cancelPromise with undefined.
350        if !self.canceled_1.get() || !self.canceled_2.get() {
351            self.cancel_promise.resolve_native(cx, &());
352        }
353    }
354}
355
356/// <https://streams.spec.whatwg.org/#readablestreamdefaultreader>
357#[dom_struct]
358pub(crate) struct ReadableStreamDefaultReader {
359    reflector_: Reflector,
360
361    /// <https://streams.spec.whatwg.org/#readablestreamgenericreader-stream>
362    stream: MutNullableDom<ReadableStream>,
363
364    read_requests: DomRefCell<VecDeque<ReadRequest>>,
365
366    /// <https://streams.spec.whatwg.org/#readablestreamgenericreader-closedpromise>
367    closed_promise: DomRefCell<TracedPromise>,
368}
369
370impl ReadableStreamDefaultReader {
371    fn new_with_proto(
372        cx: &mut JSContext,
373        global: &GlobalScope,
374        proto: Option<SafeHandleObject>,
375    ) -> DomRoot<ReadableStreamDefaultReader> {
376        let closed_promise = Promise::new(cx, global);
377        reflect_dom_object_with_proto(
378            cx,
379            Box::new(ReadableStreamDefaultReader::new_inherited(&closed_promise)),
380            global,
381            proto,
382        )
383    }
384
385    fn new_inherited(promise: &RootedPromise) -> ReadableStreamDefaultReader {
386        ReadableStreamDefaultReader {
387            reflector_: Reflector::new(),
388            stream: MutNullableDom::new(None),
389            read_requests: DomRefCell::new(Default::default()),
390            closed_promise: DomRefCell::new(promise.to_traced()),
391        }
392    }
393
394    pub(crate) fn new(
395        cx: &mut JSContext,
396        global: &GlobalScope,
397    ) -> DomRoot<ReadableStreamDefaultReader> {
398        let closed_promise = Promise::new(cx, global);
399        reflect_dom_object(cx, Box::new(Self::new_inherited(&closed_promise)), global)
400    }
401
402    /// <https://streams.spec.whatwg.org/#set-up-readable-stream-default-reader>
403    pub(crate) fn set_up(
404        &self,
405        cx: &mut JSContext,
406        stream: &ReadableStream,
407        global: &GlobalScope,
408    ) -> Fallible<()> {
409        // If ! IsReadableStreamLocked(stream) is true, throw a TypeError exception.
410        if stream.is_locked() {
411            return Err(Error::Type(c"stream is locked".to_owned()));
412        }
413        // Perform ! ReadableStreamReaderGenericInitialize(reader, stream).
414
415        self.generic_initialize(cx, global, stream);
416
417        // Set reader.[[readRequests]] to a new empty list.
418        self.read_requests.borrow_mut().clear();
419
420        Ok(())
421    }
422
423    /// <https://streams.spec.whatwg.org/#readable-stream-close>
424    pub(crate) fn close(&self, cx: &mut js::context::JSContext) {
425        // Resolve reader.[[closedPromise]] with undefined.
426        self.closed_promise.borrow().resolve_native(cx, &());
427        // If reader implements ReadableStreamDefaultReader,
428        // Let readRequests be reader.[[readRequests]].
429        rooted!(&in(cx) let mut read_requests = self.take_read_requests());
430        // Set reader.[[readRequests]] to an empty list.
431        // For each readRequest of readRequests,
432        for request in read_requests.iter() {
433            // Perform readRequest’s close steps.
434            request.close_steps(cx);
435        }
436    }
437
438    /// <https://streams.spec.whatwg.org/#readable-stream-add-read-request>
439    pub(crate) fn add_read_request(&self, read_request: &ReadRequest) {
440        self.read_requests
441            .borrow_mut()
442            .push_back(read_request.clone());
443    }
444
445    /// <https://streams.spec.whatwg.org/#readable-stream-get-num-read-requests>
446    pub(crate) fn get_num_read_requests(&self) -> usize {
447        self.read_requests.borrow().len()
448    }
449
450    /// <https://streams.spec.whatwg.org/#readable-stream-error>
451    pub(crate) fn error(&self, cx: &mut js::context::JSContext, e: SafeHandleValue) {
452        // Reject reader.[[closedPromise]] with e.
453        self.closed_promise.borrow().reject_native(cx, &e);
454
455        // Set reader.[[closedPromise]].[[PromiseIsHandled]] to true.
456        self.closed_promise.borrow().set_promise_is_handled(cx);
457
458        // Perform ! ReadableStreamDefaultReaderErrorReadRequests(reader, e).
459        self.error_read_requests(cx, e);
460    }
461
462    /// The removal steps of <https://streams.spec.whatwg.org/#readable-stream-fulfill-read-request>
463    pub(crate) fn remove_read_request(&self) -> ReadRequest {
464        self.read_requests
465            .borrow_mut()
466            .pop_front()
467            .expect("Reader must have read request when remove is called into.")
468    }
469
470    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaultreaderrelease>
471    pub(crate) fn release(&self, cx: &mut js::context::JSContext) -> Fallible<()> {
472        // Perform ! ReadableStreamReaderGenericRelease(reader).
473        self.generic_release(cx).expect("Generic release failed");
474        // Let e be a new TypeError exception.
475        rooted!(&in(cx) let mut error = UndefinedValue());
476        Error::Type(c"Reader is released".to_owned()).to_jsval(
477            cx,
478            &self.global(),
479            error.handle_mut(),
480        );
481
482        // Perform ! ReadableStreamDefaultReaderErrorReadRequests(reader, e).
483        self.error_read_requests(cx, error.handle());
484        Ok(())
485    }
486
487    fn take_read_requests(&self) -> VecDeque<ReadRequest> {
488        mem::take(&mut *self.read_requests.borrow_mut())
489    }
490
491    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaultreadererrorreadrequests>
492    fn error_read_requests(&self, cx: &mut js::context::JSContext, rval: SafeHandleValue) {
493        // step 1
494        rooted!(&in(cx) let mut read_requests = self.take_read_requests());
495
496        // step 2 & 3
497        for request in read_requests.iter() {
498            request.error_steps(cx, rval);
499        }
500    }
501
502    /// <https://streams.spec.whatwg.org/#readable-stream-default-reader-read>
503    pub(crate) fn read(&self, cx: &mut js::context::JSContext, read_request: &ReadRequest) {
504        // Let stream be reader.[[stream]].
505
506        // Assert: stream is not undefined.
507        assert!(self.stream.get().is_some());
508
509        let stream = self.stream.get().unwrap();
510
511        // Set stream.[[disturbed]] to true.
512        stream.set_is_disturbed(true);
513        // If stream.[[state]] is "closed", perform readRequest’s close steps.
514        if stream.is_closed() {
515            read_request.close_steps(cx);
516        } else if stream.is_errored() {
517            // Otherwise, if stream.[[state]] is "errored",
518            // perform readRequest’s error steps given stream.[[storedError]].
519            rooted!(&in(cx) let mut error = UndefinedValue());
520            stream.get_stored_error(error.handle_mut());
521            read_request.error_steps(cx, error.handle());
522        } else {
523            // Otherwise
524            // Assert: stream.[[state]] is "readable".
525            assert!(stream.is_readable());
526            // Perform ! stream.[[controller]].[[PullSteps]](readRequest).
527            stream.perform_pull_steps(cx, read_request);
528        }
529    }
530
531    /// Attach the byte-tee error handler to this reader's closedPromise.
532    /// Used by ReadableByteStreamTee.
533    #[allow(clippy::too_many_arguments)]
534    pub(crate) fn byte_tee_append_native_handler_to_closed_promise(
535        &self,
536        cx: &mut js::context::JSContext,
537        branch_1: &ReadableStream,
538        branch_2: &ReadableStream,
539        canceled_1: Rc<Cell<bool>>,
540        canceled_2: Rc<Cell<bool>>,
541        cancel_promise: &RootedPromise,
542        reader_version: Rc<Cell<u64>>,
543        expected_version: u64,
544    ) {
545        // Note: for byte tee we always operate on *byte controllers*.
546        let branch_1_controller = branch_1.get_byte_controller();
547        let branch_2_controller = branch_2.get_byte_controller();
548
549        let global = self.global();
550        let handler = PromiseNativeHandler::new(
551            cx,
552            &global,
553            None,
554            Some(Box::new(ByteTeeClosedPromiseRejectionHandler {
555                branch_1_controller: Dom::from_ref(&branch_1_controller),
556                branch_2_controller: Dom::from_ref(&branch_2_controller),
557                canceled_1,
558                canceled_2,
559                cancel_promise: cancel_promise.to_traced(),
560                reader_version,
561                expected_version,
562            })),
563        );
564
565        let mut realm = enter_auto_realm(cx, &*global);
566        let cx = &mut realm.current_realm();
567
568        self.closed_promise
569            .borrow()
570            .append_native_handler(cx, &handler);
571    }
572
573    /// <https://streams.spec.whatwg.org/#ref-for-readablestreamgenericreader-closedpromise%E2%91%A1>
574    pub(crate) fn default_tee_append_native_handler_to_closed_promise(
575        &self,
576        cx: &mut js::context::JSContext,
577        branch_1: &ReadableStream,
578        branch_2: &ReadableStream,
579        canceled_1: Rc<Cell<bool>>,
580        canceled_2: Rc<Cell<bool>>,
581        cancel_promise: &RootedPromise,
582    ) {
583        let branch_1_controller = branch_1.get_default_controller();
584
585        let branch_2_controller = branch_2.get_default_controller();
586
587        let global = self.global();
588        let handler = PromiseNativeHandler::new(
589            cx,
590            &global,
591            None,
592            Some(Box::new(DefaultTeeClosedPromiseRejectionHandler {
593                branch_1_controller: Dom::from_ref(&branch_1_controller),
594                branch_2_controller: Dom::from_ref(&branch_2_controller),
595                canceled_1,
596                canceled_2,
597                cancel_promise: cancel_promise.to_traced(),
598            })),
599        );
600
601        let mut realm = enter_auto_realm(cx, &*global);
602        let cx = &mut realm.current_realm();
603
604        self.closed_promise
605            .borrow()
606            .append_native_handler(cx, &handler);
607    }
608
609    /// <https://streams.spec.whatwg.org/#readablestreamdefaultreader-read-all-bytes>
610    pub(crate) fn read_all_bytes(
611        &self,
612        cx: &mut js::context::JSContext,
613        success_steps: Rc<ReadAllBytesSuccessSteps>,
614        failure_steps: Rc<ReadAllBytesFailureSteps>,
615    ) {
616        // To read all bytes from a ReadableStreamDefaultReader reader,
617        // given successSteps, which is an algorithm accepting a byte sequence,
618        // and failureSteps, which is an algorithm accepting a JavaScript value:
619        // read-loop given reader, a new byte sequence, successSteps, and failureSteps.
620        read_loop(cx, self, success_steps, failure_steps);
621    }
622
623    /// step 3 of <https://streams.spec.whatwg.org/#abstract-opdef-readablebytestreamcontrollerprocessreadrequestsusingqueue>
624    pub(crate) fn process_read_requests(
625        &self,
626        cx: &mut js::context::JSContext,
627        controller: &ReadableByteStreamController,
628    ) -> Fallible<()> {
629        // While reader.[[readRequests]] is not empty,
630        while !self.read_requests.borrow().is_empty() {
631            // If controller.[[queueTotalSize]] is 0, return.
632            if controller.get_queue_total_size() == 0.0 {
633                return Ok(());
634            }
635
636            // Let readRequest be reader.[[readRequests]][0].
637            // Remove entry from controller.[[queue]].
638            rooted!(&in(cx) let read_request = self.remove_read_request());
639
640            // Perform ! ReadableByteStreamControllerFillReadRequestFromQueue(controller, readRequest).
641            controller
642                .fill_read_request_from_queue(cx, &read_request)
643                .expect("Fill read request from queue failed");
644        }
645        Ok(())
646    }
647}
648
649impl ReadableStreamDefaultReaderMethods<crate::DomTypeHolder> for ReadableStreamDefaultReader {
650    /// <https://streams.spec.whatwg.org/#default-reader-constructor>
651    fn Constructor(
652        cx: &mut JSContext,
653        global: &GlobalScope,
654        proto: Option<SafeHandleObject>,
655        stream: &ReadableStream,
656    ) -> Fallible<DomRoot<Self>> {
657        let reader = Self::new_with_proto(cx, global, proto);
658
659        // Perform ? SetUpReadableStreamDefaultReader(this, stream).
660        reader.set_up(cx, stream, global)?;
661
662        Ok(reader)
663    }
664
665    /// <https://streams.spec.whatwg.org/#default-reader-read>
666    fn Read(&self, cx: &mut js::context::JSContext) -> RootedPromise {
667        // If this.[[stream]] is undefined, return a promise rejected with a TypeError exception.
668        if self.stream.get().is_none() {
669            rooted!(&in(cx) let mut error = UndefinedValue());
670            Error::Type(c"stream is undefined".to_owned()).to_jsval(
671                cx,
672                &self.global(),
673                error.handle_mut(),
674            );
675            return Promise::new_rejected(cx, &self.global(), error.handle());
676        }
677        // Let promise be a new promise.
678        let promise = Promise::new(cx, &self.global());
679
680        // Let readRequest be a new read request with the following items:
681        // chunk steps, given chunk
682        // Resolve promise with «[ "value" → chunk, "done" → false ]».
683        //
684        // close steps
685        // Resolve promise with «[ "value" → undefined, "done" → true ]».
686        //
687        // error steps, given e
688        // Reject promise with e.
689
690        rooted!(&in(cx) let read_request = ReadRequest::Read(promise.to_traced()));
691
692        // Perform ! ReadableStreamDefaultReaderRead(this, readRequest).
693        self.read(cx, &read_request);
694
695        // Return promise.
696        promise
697    }
698
699    /// <https://streams.spec.whatwg.org/#default-reader-release-lock>
700    fn ReleaseLock(&self, cx: &mut js::context::JSContext) -> Fallible<()> {
701        if self.stream.get().is_none() {
702            // Step 1: If this.[[stream]] is undefined, return.
703            return Ok(());
704        }
705
706        // Step 2: Perform !ReadableStreamDefaultReaderRelease(this).
707        self.release(cx)
708    }
709
710    /// <https://streams.spec.whatwg.org/#generic-reader-closed>
711    fn Closed(&self, cx: &JSContext) -> RootedPromise {
712        self.closed(cx)
713    }
714
715    /// <https://streams.spec.whatwg.org/#generic-reader-cancel>
716    fn Cancel(&self, cx: &mut js::context::JSContext, reason: SafeHandleValue) -> RootedPromise {
717        self.generic_cancel(cx, &self.global(), reason)
718    }
719}
720
721impl ReadableStreamGenericReader for ReadableStreamDefaultReader {
722    fn get_closed_promise(&self, cx: &JSContext) -> RootedPromise {
723        self.closed_promise.borrow().root(cx)
724    }
725
726    fn set_closed_promise(&self, promise: &RootedPromise) {
727        *self.closed_promise.borrow_mut() = promise.to_traced();
728    }
729
730    fn set_stream(&self, stream: Option<&ReadableStream>) {
731        self.stream.set(stream);
732    }
733
734    fn get_stream(&self) -> Option<DomRoot<ReadableStream>> {
735        self.stream.get()
736    }
737
738    fn as_default_reader(&self) -> Option<&ReadableStreamDefaultReader> {
739        Some(self)
740    }
741}