script/dom/
readablestreambyobreader.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::collections::VecDeque;
6use std::mem;
7use std::rc::Rc;
8
9use dom_struct::dom_struct;
10use js::gc::CustomAutoRooterGuard;
11use js::jsapi::Heap;
12use js::jsval::{JSVal, UndefinedValue};
13use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
14use js::typedarray::{ArrayBufferView, ArrayBufferViewU8};
15
16use super::bindings::buffer_source::HeapBufferSource;
17use super::bindings::codegen::Bindings::ReadableStreamBYOBReaderBinding::ReadableStreamBYOBReaderReadOptions;
18use super::bindings::codegen::Bindings::ReadableStreamDefaultReaderBinding::ReadableStreamReadResult;
19use super::bindings::reflector::reflect_dom_object;
20use super::readablestreamgenericreader::ReadableStreamGenericReader;
21use crate::dom::bindings::cell::DomRefCell;
22use crate::dom::bindings::codegen::Bindings::ReadableStreamBYOBReaderBinding::ReadableStreamBYOBReaderMethods;
23use crate::dom::bindings::error::{Error, ErrorToJsval, Fallible};
24use crate::dom::bindings::reflector::{DomGlobal, Reflector, reflect_dom_object_with_proto};
25use crate::dom::bindings::root::{DomRoot, MutNullableDom};
26use crate::dom::bindings::trace::RootedTraceableBox;
27use crate::dom::globalscope::GlobalScope;
28use crate::dom::promise::Promise;
29use crate::dom::readablestream::ReadableStream;
30use crate::script_runtime::{CanGc, JSContext as SafeJSContext};
31
32/// <https://streams.spec.whatwg.org/#read-into-request>
33#[derive(Clone, JSTraceable, MallocSizeOf)]
34pub enum ReadIntoRequest {
35    /// <https://streams.spec.whatwg.org/#byob-reader-read>
36    Read(#[conditional_malloc_size_of] Rc<Promise>),
37}
38
39impl ReadIntoRequest {
40    /// <https://streams.spec.whatwg.org/#ref-for-read-into-request-chunk-steps>
41    pub fn chunk_steps(&self, chunk: RootedTraceableBox<Heap<JSVal>>, can_gc: CanGc) {
42        // chunk steps, given chunk
43        // Resolve promise with «[ "value" → chunk, "done" → false ]».
44        match self {
45            ReadIntoRequest::Read(promise) => {
46                promise.resolve_native(
47                    &ReadableStreamReadResult {
48                        done: Some(false),
49                        value: chunk,
50                    },
51                    can_gc,
52                );
53            },
54        }
55    }
56
57    /// <https://streams.spec.whatwg.org/#ref-for-read-into-request-close-steps%E2%91%A0>
58    pub fn close_steps(&self, chunk: Option<RootedTraceableBox<Heap<JSVal>>>, can_gc: CanGc) {
59        // close steps, given chunk
60        // Resolve promise with «[ "value" → chunk, "done" → true ]».
61        match self {
62            ReadIntoRequest::Read(promise) => match chunk {
63                Some(chunk) => promise.resolve_native(
64                    &ReadableStreamReadResult {
65                        done: Some(true),
66                        value: chunk,
67                    },
68                    can_gc,
69                ),
70                None => {
71                    let result = RootedTraceableBox::new(Heap::default());
72                    result.set(UndefinedValue());
73                    promise.resolve_native(
74                        &ReadableStreamReadResult {
75                            done: Some(true),
76                            value: result,
77                        },
78                        can_gc,
79                    );
80                },
81            },
82        }
83    }
84
85    /// <https://streams.spec.whatwg.org/#ref-for-read-into-request-error-steps>
86    pub(crate) fn error_steps(&self, e: SafeHandleValue, can_gc: CanGc) {
87        // error steps, given e
88        // Reject promise with e.
89        match self {
90            ReadIntoRequest::Read(promise) => promise.reject_native(&e, can_gc),
91        }
92    }
93}
94
95/// <https://streams.spec.whatwg.org/#readablestreambyobreader>
96#[dom_struct]
97pub(crate) struct ReadableStreamBYOBReader {
98    reflector_: Reflector,
99
100    /// <https://streams.spec.whatwg.org/#readablestreamgenericreader-stream>
101    stream: MutNullableDom<ReadableStream>,
102
103    read_into_requests: DomRefCell<VecDeque<ReadIntoRequest>>,
104
105    /// <https://streams.spec.whatwg.org/#readablestreamgenericreader-closedpromise>
106    #[conditional_malloc_size_of]
107    closed_promise: DomRefCell<Rc<Promise>>,
108}
109
110impl ReadableStreamBYOBReader {
111    fn new_with_proto(
112        global: &GlobalScope,
113        proto: Option<SafeHandleObject>,
114        can_gc: CanGc,
115    ) -> DomRoot<ReadableStreamBYOBReader> {
116        reflect_dom_object_with_proto(
117            Box::new(ReadableStreamBYOBReader::new_inherited(global, can_gc)),
118            global,
119            proto,
120            can_gc,
121        )
122    }
123
124    fn new_inherited(global: &GlobalScope, can_gc: CanGc) -> ReadableStreamBYOBReader {
125        ReadableStreamBYOBReader {
126            reflector_: Reflector::new(),
127            stream: MutNullableDom::new(None),
128            read_into_requests: DomRefCell::new(Default::default()),
129            closed_promise: DomRefCell::new(Promise::new(global, can_gc)),
130        }
131    }
132
133    pub(crate) fn new(global: &GlobalScope, can_gc: CanGc) -> DomRoot<ReadableStreamBYOBReader> {
134        reflect_dom_object(
135            Box::new(Self::new_inherited(global, can_gc)),
136            global,
137            can_gc,
138        )
139    }
140
141    /// <https://streams.spec.whatwg.org/#set-up-readable-stream-byob-reader>
142    pub(crate) fn set_up(
143        &self,
144        stream: &ReadableStream,
145        global: &GlobalScope,
146        can_gc: CanGc,
147    ) -> Fallible<()> {
148        // If ! IsReadableStreamLocked(stream) is true, throw a TypeError exception.
149        if stream.is_locked() {
150            return Err(Error::Type("stream is locked".to_owned()));
151        }
152
153        // If stream.[[controller]] does not implement ReadableByteStreamController, throw a TypeError exception.
154        if !stream.has_byte_controller() {
155            return Err(Error::Type(
156                "stream controller is not a byte stream controller".to_owned(),
157            ));
158        }
159
160        // Perform ! ReadableStreamReaderGenericInitialize(reader, stream).
161        self.generic_initialize(global, stream, can_gc);
162
163        // Set reader.[[readIntoRequests]] to a new empty list.
164        self.read_into_requests.borrow_mut().clear();
165
166        Ok(())
167    }
168
169    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreambyobreaderrelease>
170    pub(crate) fn release(&self, can_gc: CanGc) -> Fallible<()> {
171        // Perform ! ReadableStreamReaderGenericRelease(reader).
172        self.generic_release(can_gc)
173            .expect("Generic release failed");
174        // Let e be a new TypeError exception.
175        let cx = GlobalScope::get_cx();
176        rooted!(in(*cx) let mut error = UndefinedValue());
177        Error::Type("Reader is released".to_owned()).to_jsval(
178            cx,
179            &self.global(),
180            error.handle_mut(),
181            can_gc,
182        );
183
184        // Perform ! ReadableStreamBYOBReaderErrorReadIntoRequests(reader, e).
185        self.error_read_into_requests(error.handle(), can_gc);
186        Ok(())
187    }
188
189    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreambyobreadererrorreadintorequests>
190    pub(crate) fn error_read_into_requests(&self, e: SafeHandleValue, can_gc: CanGc) {
191        // Reject reader.[[closedPromise]] with e.
192        self.closed_promise.borrow().reject_native(&e, can_gc);
193
194        // Set reader.[[closedPromise]].[[PromiseIsHandled]] to true.
195        self.closed_promise.borrow().set_promise_is_handled();
196
197        // Let readRequests be reader.[[readRequests]].
198        let mut read_into_requests = self.take_read_into_requests();
199
200        // Set reader.[[readIntoRequests]] to a new empty list.
201        for request in read_into_requests.drain(0..) {
202            // Perform readIntoRequest’s error steps, given e.
203            request.error_steps(e, can_gc);
204        }
205    }
206
207    fn take_read_into_requests(&self) -> VecDeque<ReadIntoRequest> {
208        mem::take(&mut *self.read_into_requests.borrow_mut())
209    }
210
211    /// <https://streams.spec.whatwg.org/#readable-stream-add-read-into-request>
212    pub(crate) fn add_read_into_request(&self, read_request: &ReadIntoRequest) {
213        self.read_into_requests
214            .borrow_mut()
215            .push_back(read_request.clone());
216    }
217
218    /// <https://streams.spec.whatwg.org/#readable-stream-cancel>
219    pub(crate) fn cancel(&self, can_gc: CanGc) {
220        // If reader is not undefined and reader implements ReadableStreamBYOBReader,
221        // Let readIntoRequests be reader.[[readIntoRequests]].
222        let mut read_into_requests = self.take_read_into_requests();
223        // Set reader.[[readIntoRequests]] to an empty list.
224        // Perform readIntoRequest’s close steps, given undefined.
225        for request in read_into_requests.drain(0..) {
226            // Perform readIntoRequest’s close steps, given undefined.
227            request.close_steps(None, can_gc);
228        }
229    }
230
231    pub(crate) fn close(&self, can_gc: CanGc) {
232        // Resolve reader.[[closedPromise]] with undefined.
233        self.closed_promise.borrow().resolve_native(&(), can_gc);
234    }
235
236    /// <https://streams.spec.whatwg.org/#readable-stream-byob-reader-read>
237    pub(crate) fn read(
238        &self,
239        cx: SafeJSContext,
240        view: HeapBufferSource<ArrayBufferViewU8>,
241        options: &ReadableStreamBYOBReaderReadOptions,
242        read_into_request: &ReadIntoRequest,
243        can_gc: CanGc,
244    ) {
245        // Let stream be reader.[[stream]].
246
247        // Assert: stream is not undefined.
248        assert!(self.stream.get().is_some());
249
250        let stream = self.stream.get().unwrap();
251
252        // Set stream.[[disturbed]] to true.
253        stream.set_is_disturbed(true);
254        // If stream.[[state]] is "errored", perform readIntoRequest’s error steps given stream.[[storedError]].
255        if stream.is_errored() {
256            let cx = GlobalScope::get_cx();
257            rooted!(in(*cx) let mut error = UndefinedValue());
258            stream.get_stored_error(error.handle_mut());
259
260            read_into_request.error_steps(error.handle(), can_gc);
261        } else {
262            // Otherwise,
263            // perform ! ReadableByteStreamControllerPullInto(stream.[[controller]], view, min, readIntoRequest).
264            stream.perform_pull_into(cx, read_into_request, view, options, can_gc);
265        }
266    }
267
268    pub(crate) fn get_num_read_into_requests(&self) -> usize {
269        self.read_into_requests.borrow().len()
270    }
271
272    pub(crate) fn remove_read_into_request(&self) -> ReadIntoRequest {
273        self.read_into_requests
274            .borrow_mut()
275            .pop_front()
276            .expect("read into requests is empty")
277    }
278}
279
280impl ReadableStreamBYOBReaderMethods<crate::DomTypeHolder> for ReadableStreamBYOBReader {
281    /// <https://streams.spec.whatwg.org/#byob-reader-constructor>
282    fn Constructor(
283        global: &GlobalScope,
284        proto: Option<SafeHandleObject>,
285        can_gc: CanGc,
286        stream: &ReadableStream,
287    ) -> Fallible<DomRoot<Self>> {
288        let reader = Self::new_with_proto(global, proto, can_gc);
289
290        // Perform ? SetUpReadableStreamBYOBReader(this, stream).
291        Self::set_up(&reader, stream, global, can_gc)?;
292
293        Ok(reader)
294    }
295
296    /// <https://streams.spec.whatwg.org/#byob-reader-read>
297    fn Read(
298        &self,
299        view: CustomAutoRooterGuard<ArrayBufferView>,
300        options: &ReadableStreamBYOBReaderReadOptions,
301        can_gc: CanGc,
302    ) -> Rc<Promise> {
303        let view = HeapBufferSource::<ArrayBufferViewU8>::from_view(view);
304
305        // Let promise be a new promise.
306        let promise = Promise::new(&self.global(), can_gc);
307
308        let cx = GlobalScope::get_cx();
309        // If view.[[ByteLength]] is 0, return a promise rejected with a TypeError exception.
310        if view.byte_length() == 0 {
311            promise.reject_error(Error::Type("view byte length is 0".to_owned()), can_gc);
312            return promise;
313        }
314        // If view.[[ViewedArrayBuffer]].[[ArrayBufferByteLength]] is 0,
315        // return a promise rejected with a TypeError exception.
316        if view.viewed_buffer_array_byte_length(cx) == 0 {
317            promise.reject_error(
318                Error::Type("viewed buffer byte length is 0".to_owned()),
319                can_gc,
320            );
321            return promise;
322        }
323
324        // If ! IsDetachedBuffer(view.[[ViewedArrayBuffer]]) is true,
325        // return a promise rejected with a TypeError exception.
326        if view.is_detached_buffer(cx) {
327            promise.reject_error(Error::Type("view is detached".to_owned()), can_gc);
328            return promise;
329        }
330
331        // If options["min"] is 0, return a promise rejected with a TypeError exception.
332        if options.min == 0 {
333            promise.reject_error(Error::Type("min is 0".to_owned()), can_gc);
334            return promise;
335        }
336
337        // If view has a [[TypedArrayName]] internal slot,
338        if view.has_typed_array_name() {
339            // If options["min"] > view.[[ArrayLength]], return a promise rejected with a RangeError exception.
340            if options.min > (view.get_typed_array_length() as u64) {
341                promise.reject_error(
342                    Error::Range("min is greater than array length".to_owned()),
343                    can_gc,
344                );
345                return promise;
346            }
347        } else {
348            // Otherwise (i.e., it is a DataView),
349            // If options["min"] > view.[[ByteLength]], return a promise rejected with a RangeError exception.
350            if options.min > (view.byte_length() as u64) {
351                promise.reject_error(
352                    Error::Range("min is greater than byte length".to_owned()),
353                    can_gc,
354                );
355                return promise;
356            }
357        }
358
359        // If this.[[stream]] is undefined, return a promise rejected with a TypeError exception.
360        if self.stream.get().is_none() {
361            promise.reject_error(
362                Error::Type("min is greater than byte length".to_owned()),
363                can_gc,
364            );
365            return promise;
366        }
367
368        // Let readIntoRequest be a new read-into request with the following items:
369        //
370        // chunk steps, given chunk
371        // Resolve promise with «[ "value" → chunk, "done" → false ]».
372        //
373        // close steps, given chunk
374        // Resolve promise with «[ "value" → chunk, "done" → true ]».
375        //
376        // error steps, given e
377        // Reject promise with e
378        let read_into_request = ReadIntoRequest::Read(promise.clone());
379
380        // Perform ! ReadableStreamBYOBReaderRead(this, view, options["min"], readIntoRequest).
381        self.read(cx, view, options, &read_into_request, can_gc);
382
383        // Return promise.
384        promise
385    }
386
387    /// <https://streams.spec.whatwg.org/#byob-reader-release-lock>
388    fn ReleaseLock(&self, can_gc: CanGc) -> Fallible<()> {
389        if self.stream.get().is_none() {
390            // If this.[[stream]] is undefined, return.
391            return Ok(());
392        }
393
394        // Perform !ReadableStreamBYOBReaderRelease(this).
395        self.release(can_gc)
396    }
397
398    /// <https://streams.spec.whatwg.org/#generic-reader-closed>
399    fn Closed(&self) -> Rc<Promise> {
400        self.closed()
401    }
402
403    /// <https://streams.spec.whatwg.org/#generic-reader-cancel>
404    fn Cancel(&self, cx: SafeJSContext, reason: SafeHandleValue, can_gc: CanGc) -> Rc<Promise> {
405        self.generic_cancel(cx, &self.global(), reason, can_gc)
406    }
407}
408
409impl ReadableStreamGenericReader for ReadableStreamBYOBReader {
410    fn get_closed_promise(&self) -> Rc<Promise> {
411        self.closed_promise.borrow().clone()
412    }
413
414    fn set_closed_promise(&self, promise: Rc<Promise>) {
415        *self.closed_promise.borrow_mut() = promise;
416    }
417
418    fn set_stream(&self, stream: Option<&ReadableStream>) {
419        self.stream.set(stream);
420    }
421
422    fn get_stream(&self) -> Option<DomRoot<ReadableStream>> {
423        self.stream.get()
424    }
425
426    fn as_byob_reader(&self) -> Option<&ReadableStreamBYOBReader> {
427        Some(self)
428    }
429}
script/dom/readablestreambyobreader.rs

script/dom/
readablestreambyobreader.rs
