script: Implement set_up, ReleaseLock, Closed, Cancel, acquire_byob_reader for ReadableStreamBYOBReader (#34958)

* introduce ReadableStreamGenericReader to share functionality between ReadableStreamDefaultReader and ReadableStreamBYOBReader

Signed-off-by: Taym Haddadi <haddadi.taym@gmail.com>

* implement set_up, ReleaseLock, Closed, Cancel for ReadableStreamBYOBReader

Signed-off-by: Taym Haddadi <haddadi.taym@gmail.com>

* Fix clippy

Signed-off-by: Taym Haddadi <haddadi.taym@gmail.com>

* Implement acquire_byob_reader

Signed-off-by: Taym Haddadi <haddadi.taym@gmail.com>

* Remove #[allow(crown::unrooted_must_root)] from ReadRequest and ReadIntoRequest

Signed-off-by: Taym Haddadi <haddadi.taym@gmail.com>

* Fix clippy

Signed-off-by: Taym Haddadi <haddadi.taym@gmail.com>

---------

Signed-off-by: Taym Haddadi <haddadi.taym@gmail.com>
This commit is contained in:
Taym Haddadi 2025-01-15 23:34:00 +01:00 committed by GitHub
parent 05a13df294
commit 3225d196fa
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 466 additions and 196 deletions

View file

@ -547,7 +547,7 @@ DOMInterfaces = {
},
"ReadableStreamBYOBReader": {
"canGc": ["Read", "Closed", "Cancel"]
"canGc": ["Read", "Cancel"]
},
"ReadableStreamDefaultReader": {

View file

@ -501,6 +501,7 @@ pub(crate) mod readablestreambyobreader;
pub(crate) mod readablestreambyobrequest;
pub(crate) mod readablestreamdefaultcontroller;
pub(crate) mod readablestreamdefaultreader;
pub(crate) mod readablestreamgenericreader;
pub(crate) mod request;
pub(crate) mod resizeobserver;
pub(crate) mod resizeobserverentry;

View file

@ -26,7 +26,7 @@ use crate::dom::bindings::conversions::{ConversionBehavior, ConversionResult};
use crate::dom::bindings::error::Error;
use crate::dom::bindings::import::module::Fallible;
use crate::dom::bindings::import::module::UnionTypes::ReadableStreamDefaultReaderOrReadableStreamBYOBReader as ReadableStreamReader;
use crate::dom::bindings::reflector::{DomObject, Reflector, reflect_dom_object, reflect_dom_object_with_proto};
use crate::dom::bindings::reflector::{DomObject, Reflector, reflect_dom_object_with_proto};
use crate::dom::bindings::root::{DomRoot, MutNullableDom, Dom};
use crate::dom::bindings::trace::RootedTraceableBox;
use crate::dom::bindings::utils::get_dictionary_property;
@ -277,12 +277,12 @@ impl ReadableStream {
}
/// Call into the release steps of the controller,
pub(crate) fn perform_release_steps(&self) {
match self.controller {
ControllerType::Default(ref controller) => controller
pub(crate) fn perform_release_steps(&self) -> Fallible<()> {
match &self.controller {
ControllerType::Default(controller) => controller
.get()
.expect("Stream should have controller.")
.perform_release_steps(),
.map(|controller_ref| controller_ref.perform_release_steps())
.unwrap_or_else(|| Err(Error::Type("Stream should have controller.".to_string()))),
ControllerType::Byte(_) => todo!(),
}
}
@ -426,14 +426,7 @@ impl ReadableStream {
can_gc: CanGc,
) -> Fallible<DomRoot<ReadableStreamDefaultReader>> {
// Let reader be a new ReadableStreamDefaultReader.
let reader = reflect_dom_object(
Box::new(ReadableStreamDefaultReader::new_inherited(
&self.global(),
can_gc,
)),
&*self.global(),
can_gc,
);
let reader = ReadableStreamDefaultReader::new(&self.global(), can_gc);
// Perform ? SetUpReadableStreamDefaultReader(reader, stream).
reader.set_up(self, &self.global(), can_gc)?;
@ -442,6 +435,20 @@ impl ReadableStream {
Ok(reader)
}
/// <https://streams.spec.whatwg.org/#acquire-readable-stream-byob-reader>
pub(crate) fn acquire_byob_reader(
&self,
can_gc: CanGc,
) -> Fallible<DomRoot<ReadableStreamBYOBReader>> {
// Let reader be a new ReadableStreamBYOBReader.
let reader = ReadableStreamBYOBReader::new(&self.global(), can_gc);
// Perform ? SetUpReadableStreamBYOBReader(reader, stream).
reader.set_up(self, &self.global(), can_gc)?;
// Return reader.
Ok(reader)
}
pub(crate) fn get_default_controller(&self) -> DomRoot<ReadableStreamDefaultController> {
match self.controller {
ControllerType::Default(ref controller) => {
@ -484,7 +491,7 @@ impl ReadableStream {
let Some(reader) = reader.get() else {
unreachable!("Attempt to stop reading without having first acquired a reader.");
};
reader.release();
reader.release().expect("Reader release cannot fail.");
},
ReaderType::BYOB(_) => {
unreachable!("Native stop reading can only be done with a default reader.")
@ -527,6 +534,10 @@ impl ReadableStream {
}
}
pub(crate) fn has_byte_controller(&self) -> bool {
matches!(self.controller, ControllerType::Byte(_))
}
/// <https://streams.spec.whatwg.org/#readable-stream-get-num-read-requests>
pub(crate) fn get_num_read_requests(&self) -> usize {
assert!(self.has_default_reader());
@ -592,7 +603,7 @@ impl ReadableStream {
// step 5 & 6
reader.close();
},
ReaderType::BYOB(ref _reader) => todo!(),
ReaderType::BYOB(ref _reader) => {},
}
}
@ -621,8 +632,17 @@ impl ReadableStream {
}
// Perform ! ReadableStreamClose(stream).
self.close();
// step 5, 6, 7, 8
// TODO: run the bytes reader steps.
// If reader is not undefined and reader implements ReadableStreamBYOBReader,
match self.reader {
ReaderType::BYOB(ref reader) => {
if let Some(reader) = reader.get() {
// step 6.1, 6.2 & 6.3 of https://streams.spec.whatwg.org/#readable-stream-cancel
reader.close();
}
},
ReaderType::Default(ref _reader) => {},
}
// Let sourceCancelPromise be ! stream.[[controller]].[[CancelSteps]](reason).
let source_cancel_promise = match self.controller {
@ -656,16 +676,27 @@ impl ReadableStream {
result_promise
}
pub(crate) fn set_reader(&self, new_reader: Option<&ReadableStreamDefaultReader>) {
match self.reader {
ReaderType::Default(ref reader) => {
reader.set(new_reader);
#[allow(crown::unrooted_must_root)]
pub(crate) fn set_reader(&self, new_reader: Option<ReaderType>) {
match (&self.reader, new_reader) {
(ReaderType::Default(ref reader), Some(ReaderType::Default(new_reader))) => {
reader.set(new_reader.get().as_deref());
},
ReaderType::BYOB(_) => {
unreachable!("Setting a reader can only be done on a default reader.")
(ReaderType::BYOB(ref reader), Some(ReaderType::BYOB(new_reader))) => {
reader.set(new_reader.get().as_deref());
},
(ReaderType::Default(ref reader), None) => {
reader.set(None);
},
(ReaderType::BYOB(ref reader), None) => {
reader.set(None);
},
(_, _) => {
unreachable!("Setting a mismatched reader type is not allowed.");
},
}
}
/// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaulttee>
#[allow(crown::unrooted_must_root)]
fn default_tee(
@ -680,7 +711,9 @@ impl ReadableStream {
// Let reader be ? AcquireReadableStreamDefaultReader(stream).
let reader = self.acquire_default_reader(can_gc)?;
self.set_reader(Some(&reader));
self.set_reader(Some(ReaderType::Default(MutNullableDom::new(Some(
&reader,
)))));
// Let reading be false.
let reading = Rc::new(Cell::new(false));
@ -874,7 +907,7 @@ impl ReadableStreamMethods<crate::DomTypeHolder> for ReadableStream {
if self.is_locked() {
// If ! IsReadableStreamLocked(this) is true,
// return a promise rejected with a TypeError exception.
let promise = Promise::new(&self.reflector_.global(), can_gc);
let promise = Promise::new(&self.global(), can_gc);
promise.reject_error(Error::Type("stream is not locked".to_owned()));
promise
} else {
@ -899,8 +932,8 @@ impl ReadableStreamMethods<crate::DomTypeHolder> for ReadableStream {
assert!(options.mode.unwrap() == ReadableStreamReaderMode::Byob);
// 3. Return ? AcquireReadableStreamBYOBReader(this).
Err(Error::Type(
"AcquireReadableStreamBYOBReader is not implemented".to_owned(),
Ok(ReadableStreamReader::ReadableStreamBYOBReader(
self.acquire_byob_reader(can_gc)?,
))
}

View file

@ -1,58 +1,196 @@
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
#![allow(dead_code)]
use std::collections::VecDeque;
use std::mem;
use std::rc::Rc;
use dom_struct::dom_struct;
use js::gc::CustomAutoRooterGuard;
use js::jsapi::Heap;
use js::jsval::{JSVal, UndefinedValue};
use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
use js::typedarray::ArrayBufferView;
use super::bindings::reflector::reflect_dom_object;
use super::readablestreamgenericreader::ReadableStreamGenericReader;
use crate::dom::bindings::cell::DomRefCell;
use crate::dom::bindings::codegen::Bindings::ReadableStreamBYOBReaderBinding::ReadableStreamBYOBReaderMethods;
use crate::dom::bindings::error::Error;
use crate::dom::bindings::import::module::Fallible;
use crate::dom::bindings::reflector::{reflect_dom_object, DomObject, Reflector};
use crate::dom::bindings::root::DomRoot;
use crate::dom::bindings::reflector::{reflect_dom_object_with_proto, DomObject, Reflector};
use crate::dom::bindings::root::{DomRoot, MutNullableDom};
use crate::dom::bindings::trace::RootedTraceableBox;
use crate::dom::globalscope::GlobalScope;
use crate::dom::promise::Promise;
use crate::dom::readablestream::ReadableStream;
use crate::script_runtime::{CanGc, JSContext as SafeJSContext};
/// <https://streams.spec.whatwg.org/#read-into-request>
#[derive(JSTraceable, MallocSizeOf)]
pub enum ReadIntoRequest {
/// <https://streams.spec.whatwg.org/#byob-reader-read>
Read(#[ignore_malloc_size_of = "Rc is hard"] Rc<Promise>),
}
impl ReadIntoRequest {
/// <https://streams.spec.whatwg.org/#read-into-request-chunk-steps>
pub fn chunk_steps(&self, _chunk: RootedTraceableBox<Heap<JSVal>>) {
todo!()
}
/// <https://streams.spec.whatwg.org/#read-into-request-close-steps>
pub fn close_steps(&self, _chunk: Option<RootedTraceableBox<Heap<JSVal>>>) {
todo!()
}
/// <https://streams.spec.whatwg.org/#read-into-request-error-steps>
pub(crate) fn error_steps(&self, _e: SafeHandleValue) {
todo!()
}
}
/// <https://streams.spec.whatwg.org/#readablestreambyobreader>
#[dom_struct]
pub(crate) struct ReadableStreamBYOBReader {
reflector_: Reflector,
/// <https://streams.spec.whatwg.org/#readablestreamgenericreader-stream>
stream: MutNullableDom<ReadableStream>,
read_into_requests: DomRefCell<VecDeque<ReadIntoRequest>>,
/// <https://streams.spec.whatwg.org/#readablestreamgenericreader-closedpromise>
#[ignore_malloc_size_of = "Rc is hard"]
closed_promise: DomRefCell<Rc<Promise>>,
}
impl ReadableStreamBYOBReader {
fn new_inherited() -> ReadableStreamBYOBReader {
fn new_with_proto(
global: &GlobalScope,
proto: Option<SafeHandleObject>,
can_gc: CanGc,
) -> DomRoot<ReadableStreamBYOBReader> {
reflect_dom_object_with_proto(
Box::new(ReadableStreamBYOBReader::new_inherited(global, can_gc)),
global,
proto,
can_gc,
)
}
fn new_inherited(global: &GlobalScope, can_gc: CanGc) -> ReadableStreamBYOBReader {
ReadableStreamBYOBReader {
reflector_: Reflector::new(),
stream: MutNullableDom::new(None),
read_into_requests: DomRefCell::new(Default::default()),
closed_promise: DomRefCell::new(Promise::new(global, can_gc)),
}
}
fn new(global: &GlobalScope, can_gc: CanGc) -> DomRoot<ReadableStreamBYOBReader> {
pub(crate) fn new(global: &GlobalScope, can_gc: CanGc) -> DomRoot<ReadableStreamBYOBReader> {
reflect_dom_object(
Box::new(ReadableStreamBYOBReader::new_inherited()),
Box::new(Self::new_inherited(global, can_gc)),
global,
can_gc,
)
}
/// <https://streams.spec.whatwg.org/#set-up-readable-stream-byob-reader>
pub(crate) fn set_up(
&self,
stream: &ReadableStream,
global: &GlobalScope,
can_gc: CanGc,
) -> Fallible<()> {
// If ! IsReadableStreamLocked(stream) is true, throw a TypeError exception.
if stream.is_locked() {
return Err(Error::Type("stream is locked".to_owned()));
}
// If stream.[[controller]] does not implement ReadableByteStreamController, throw a TypeError exception.
if !stream.has_byte_controller() {
return Err(Error::Type(
"stream controller is not a byte stream controller".to_owned(),
));
}
// Perform ! ReadableStreamReaderGenericInitialize(reader, stream).
self.generic_initialize(global, stream, can_gc)?;
// Set reader.[[readIntoRequests]] to a new empty list.
self.read_into_requests.borrow_mut().clear();
Ok(())
}
/// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreambyobreaderrelease>
#[allow(unsafe_code)]
pub(crate) fn release(&self) -> Fallible<()> {
// Perform ! ReadableStreamReaderGenericRelease(reader).
self.generic_release()?;
// Let e be a new TypeError exception.
let cx = GlobalScope::get_cx();
rooted!(in(*cx) let mut error = UndefinedValue());
unsafe {
Error::Type("Reader is released".to_owned()).to_jsval(
*cx,
&self.global(),
error.handle_mut(),
)
};
// Perform ! ReadableStreamBYOBReaderErrorReadIntoRequests(reader, e).
self.error_read_into_requests(error.handle());
Ok(())
}
/// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreambyobreadererrorreadintorequests>
fn error_read_into_requests(&self, rval: SafeHandleValue) {
// Let readRequests be reader.[[readRequests]].
let mut read_into_requests = self.take_read_into_requests();
// Set reader.[[readIntoRequests]] to a new empty list.
for request in read_into_requests.drain(0..) {
// Perform readIntoRequests error steps, given e.
request.error_steps(rval);
}
}
fn take_read_into_requests(&self) -> VecDeque<ReadIntoRequest> {
mem::take(&mut *self.read_into_requests.borrow_mut())
}
/// <https://streams.spec.whatwg.org/#readable-stream-cancel>
pub(crate) fn close(&self) {
// If reader is not undefined and reader implements ReadableStreamBYOBReader,
// Let readIntoRequests be reader.[[readIntoRequests]].
let mut read_into_requests = self.take_read_into_requests();
// Set reader.[[readIntoRequests]] to an empty list.
// Perform readIntoRequests close steps, given undefined.
for request in read_into_requests.drain(0..) {
// Perform readIntoRequests close steps, given undefined.
request.close_steps(None);
}
}
}
impl ReadableStreamBYOBReaderMethods<crate::DomTypeHolder> for ReadableStreamBYOBReader {
/// <https://streams.spec.whatwg.org/#byob-reader-constructor>
fn Constructor(
_global: &GlobalScope,
_proto: Option<SafeHandleObject>,
_can_gc: CanGc,
_stream: &ReadableStream,
global: &GlobalScope,
proto: Option<SafeHandleObject>,
can_gc: CanGc,
stream: &ReadableStream,
) -> Fallible<DomRoot<Self>> {
// TODO
Err(Error::NotFound)
let reader = Self::new_with_proto(global, proto, can_gc);
// Perform ? SetUpReadableStreamBYOBReader(this, stream).
Self::set_up(&reader, stream, global, can_gc)?;
Ok(reader)
}
/// <https://streams.spec.whatwg.org/#byob-reader-read>
@ -63,19 +201,44 @@ impl ReadableStreamBYOBReaderMethods<crate::DomTypeHolder> for ReadableStreamBYO
/// <https://streams.spec.whatwg.org/#byob-reader-release-lock>
fn ReleaseLock(&self) -> Fallible<()> {
// TODO
Err(Error::NotFound)
if self.stream.get().is_none() {
// If this.[[stream]] is undefined, return.
return Ok(());
}
// Perform !ReadableStreamBYOBReaderRelease(this).
self.release()
}
/// <https://streams.spec.whatwg.org/#generic-reader-closed>
fn Closed(&self, can_gc: CanGc) -> Rc<Promise> {
// TODO
Promise::new(&self.reflector_.global(), can_gc)
fn Closed(&self) -> Rc<Promise> {
self.closed()
}
/// <https://streams.spec.whatwg.org/#generic-reader-cancel>
fn Cancel(&self, _cx: SafeJSContext, _reason: SafeHandleValue, can_gc: CanGc) -> Rc<Promise> {
// TODO
Promise::new(&self.reflector_.global(), can_gc)
fn Cancel(&self, _cx: SafeJSContext, reason: SafeHandleValue, can_gc: CanGc) -> Rc<Promise> {
self.cancel(&self.global(), reason, can_gc)
}
}
impl ReadableStreamGenericReader for ReadableStreamBYOBReader {
fn get_closed_promise(&self) -> Rc<Promise> {
self.closed_promise.borrow().clone()
}
fn set_closed_promise(&self, promise: Rc<Promise>) {
*self.closed_promise.borrow_mut() = promise;
}
fn set_stream(&self, stream: Option<&ReadableStream>) {
self.stream.set(stream);
}
fn get_stream(&self) -> Option<DomRoot<ReadableStream>> {
self.stream.get()
}
fn as_byob_reader(&self) -> Option<&ReadableStreamBYOBReader> {
Some(self)
}
}

View file

@ -612,8 +612,9 @@ impl ReadableStreamDefaultController {
}
/// <https://streams.spec.whatwg.org/#ref-for-abstract-opdef-readablestreamcontroller-releasesteps>
pub(crate) fn perform_release_steps(&self) {
pub(crate) fn perform_release_steps(&self) -> Fallible<()> {
// step 1 - Return.
Ok(())
}
/// <https://streams.spec.whatwg.org/#readable-stream-default-controller-enqueue>

View file

@ -12,6 +12,7 @@ use js::jsapi::Heap;
use js::jsval::{JSVal, UndefinedValue};
use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
use super::bindings::reflector::reflect_dom_object;
use super::bindings::root::MutNullableDom;
use super::types::ReadableStreamDefaultController;
use crate::dom::bindings::cell::DomRefCell;
@ -28,15 +29,15 @@ use crate::dom::globalscope::GlobalScope;
use crate::dom::promise::Promise;
use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
use crate::dom::readablestream::ReadableStream;
use crate::dom::readablestreamgenericreader::ReadableStreamGenericReader;
use crate::realms::{enter_realm, InRealm};
use crate::script_runtime::{CanGc, JSContext as SafeJSContext};
/// <https://streams.spec.whatwg.org/#read-request>
#[derive(Clone, JSTraceable)]
#[crown::unrooted_must_root_lint::must_root]
#[derive(Clone, JSTraceable, MallocSizeOf)]
pub(crate) enum ReadRequest {
/// <https://streams.spec.whatwg.org/#default-reader-read>
Read(Rc<Promise>),
Read(#[ignore_malloc_size_of = "Rc is hard"] Rc<Promise>),
/// <https://streams.spec.whatwg.org/#ref-for-read-request%E2%91%A2>
DefaultTee {
tee_read_request: Dom<DefaultTeeReadRequest>,
@ -129,7 +130,6 @@ pub(crate) struct ReadableStreamDefaultReader {
/// <https://streams.spec.whatwg.org/#readablestreamgenericreader-stream>
stream: MutNullableDom<ReadableStream>,
#[ignore_malloc_size_of = "no VecDeque support"]
read_requests: DomRefCell<VecDeque<ReadRequest>>,
/// <https://streams.spec.whatwg.org/#readablestreamgenericreader-closedpromise>
@ -138,22 +138,6 @@ pub(crate) struct ReadableStreamDefaultReader {
}
impl ReadableStreamDefaultReader {
/// <https://streams.spec.whatwg.org/#default-reader-constructor>
#[allow(non_snake_case)]
pub(crate) fn Constructor(
global: &GlobalScope,
proto: Option<SafeHandleObject>,
can_gc: CanGc,
stream: &ReadableStream,
) -> Fallible<DomRoot<Self>> {
let reader = Self::new_with_proto(global, proto, can_gc);
// Perform ? SetUpReadableStreamDefaultReader(this, stream).
Self::set_up(&reader, stream, global, can_gc)?;
Ok(reader)
}
fn new_with_proto(
global: &GlobalScope,
proto: Option<SafeHandleObject>,
@ -167,10 +151,7 @@ impl ReadableStreamDefaultReader {
)
}
pub(crate) fn new_inherited(
global: &GlobalScope,
can_gc: CanGc,
) -> ReadableStreamDefaultReader {
fn new_inherited(global: &GlobalScope, can_gc: CanGc) -> ReadableStreamDefaultReader {
ReadableStreamDefaultReader {
reflector_: Reflector::new(),
stream: MutNullableDom::new(None),
@ -179,6 +160,14 @@ impl ReadableStreamDefaultReader {
}
}
pub(crate) fn new(global: &GlobalScope, can_gc: CanGc) -> DomRoot<ReadableStreamDefaultReader> {
reflect_dom_object(
Box::new(Self::new_inherited(global, can_gc)),
global,
can_gc,
)
}
/// <https://streams.spec.whatwg.org/#set-up-readable-stream-default-reader>
pub(crate) fn set_up(
&self,
@ -200,48 +189,7 @@ impl ReadableStreamDefaultReader {
Ok(())
}
/// <https://streams.spec.whatwg.org/#readable-stream-reader-generic-initialize>
pub(crate) fn generic_initialize(
&self,
global: &GlobalScope,
stream: &ReadableStream,
can_gc: CanGc,
) -> Fallible<()> {
// Set reader.[[stream]] to stream.
self.stream.set(Some(stream));
// Set stream.[[reader]] to reader.
stream.set_reader(Some(self));
if stream.is_readable() {
// If stream.[[state]] is "readable
// Set reader.[[closedPromise]] to a new promise.
*self.closed_promise.borrow_mut() = Promise::new(global, can_gc);
} else if stream.is_closed() {
// Otherwise, if stream.[[state]] is "closed",
// Set reader.[[closedPromise]] to a promise resolved with undefined.
let cx = GlobalScope::get_cx();
rooted!(in(*cx) let mut rval = UndefinedValue());
*self.closed_promise.borrow_mut() = Promise::new_resolved(global, cx, rval.handle())?;
} else {
// Assert: stream.[[state]] is "errored"
assert!(stream.is_errored());
// Set reader.[[closedPromise]] to a promise rejected with stream.[[storedError]].
let cx = GlobalScope::get_cx();
rooted!(in(*cx) let mut error = UndefinedValue());
stream.get_stored_error(error.handle_mut());
*self.closed_promise.borrow_mut() = Promise::new_rejected(global, cx, error.handle())?;
// Set reader.[[closedPromise]].[[PromiseIsHandled]] to true
self.closed_promise.borrow().set_promise_is_handled();
}
Ok(())
}
/// <https://streams.spec.whatwg.org/#readable-stream-close>
#[allow(crown::unrooted_must_root)]
pub(crate) fn close(&self) {
// Resolve reader.[[closedPromise]] with undefined.
self.closed_promise.borrow().resolve_native(&());
@ -281,7 +229,6 @@ impl ReadableStreamDefaultReader {
}
/// The removal steps of <https://streams.spec.whatwg.org/#readable-stream-fulfill-read-request>
#[allow(crown::unrooted_must_root)]
pub(crate) fn remove_read_request(&self) -> ReadRequest {
self.read_requests
.borrow_mut()
@ -289,56 +236,11 @@ impl ReadableStreamDefaultReader {
.expect("Reader must have read request when remove is called into.")
}
/// <https://streams.spec.whatwg.org/#readable-stream-reader-generic-release>
#[allow(unsafe_code)]
pub(crate) fn generic_release(&self) {
// Let stream be reader.[[stream]].
// Assert: stream is not undefined.
assert!(self.stream.get().is_some());
if let Some(stream) = self.stream.get() {
// Assert: stream.[[reader]] is reader.
assert!(stream.has_default_reader());
if stream.is_readable() {
// If stream.[[state]] is "readable", reject reader.[[closedPromise]] with a TypeError exception.
self.closed_promise
.borrow()
.reject_error(Error::Type("stream state is not readable".to_owned()));
} else {
// Otherwise, set reader.[[closedPromise]] to a promise rejected with a TypeError exception.
let cx = GlobalScope::get_cx();
rooted!(in(*cx) let mut error = UndefinedValue());
unsafe {
Error::Type("Cannot release lock due to stream state.".to_owned()).to_jsval(
*cx,
&self.global(),
error.handle_mut(),
)
};
*self.closed_promise.borrow_mut() =
Promise::new_rejected(&self.global(), cx, error.handle()).unwrap();
}
// Set reader.[[closedPromise]].[[PromiseIsHandled]] to true.
self.closed_promise.borrow().set_promise_is_handled();
// Perform ! stream.[[controller]].[[ReleaseSteps]]().
stream.perform_release_steps();
// Set stream.[[reader]] to undefined.
stream.set_reader(None);
// Set reader.[[stream]] to undefined.
self.stream.set(None);
}
}
/// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaultreaderrelease>
#[allow(unsafe_code)]
pub(crate) fn release(&self) {
pub(crate) fn release(&self) -> Fallible<()> {
// Perform ! ReadableStreamReaderGenericRelease(reader).
self.generic_release();
self.generic_release()?;
// Let e be a new TypeError exception.
let cx = GlobalScope::get_cx();
rooted!(in(*cx) let mut error = UndefinedValue());
@ -352,28 +254,14 @@ impl ReadableStreamDefaultReader {
// Perform ! ReadableStreamDefaultReaderErrorReadRequests(reader, e).
self.error_read_requests(error.handle());
Ok(())
}
#[allow(crown::unrooted_must_root)]
fn take_read_requests(&self) -> VecDeque<ReadRequest> {
mem::take(&mut *self.read_requests.borrow_mut())
}
/// <https://streams.spec.whatwg.org/#readable-stream-reader-generic-cancel>
fn generic_cancel(&self, reason: SafeHandleValue, can_gc: CanGc) -> Rc<Promise> {
// Let stream be reader.[[stream]].
let stream = self.stream.get();
// Assert: stream is not undefined.
let stream =
stream.expect("Reader should have a stream when generic cancel is called into.");
// Return ! ReadableStreamCancel(stream, reason).
stream.cancel(reason, can_gc)
}
/// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaultreadererrorreadrequests>
#[allow(crown::unrooted_must_root)]
fn error_read_requests(&self, rval: SafeHandleValue) {
// step 1
let mut read_requests = self.take_read_requests();
@ -458,12 +346,16 @@ impl ReadableStreamDefaultReaderMethods<crate::DomTypeHolder> for ReadableStream
can_gc: CanGc,
stream: &ReadableStream,
) -> Fallible<DomRoot<Self>> {
ReadableStreamDefaultReader::Constructor(global, proto, can_gc, stream)
let reader = Self::new_with_proto(global, proto, can_gc);
// Perform ? SetUpReadableStreamDefaultReader(this, stream).
Self::set_up(&reader, stream, global, can_gc)?;
Ok(reader)
}
/// <https://streams.spec.whatwg.org/#default-reader-read>
#[allow(unsafe_code)]
#[allow(crown::unrooted_must_root)]
fn Read(&self, can_gc: CanGc) -> Rc<Promise> {
// If this.[[stream]] is undefined, return a promise rejected with a TypeError exception.
if self.stream.get().is_none() {
@ -504,30 +396,45 @@ impl ReadableStreamDefaultReaderMethods<crate::DomTypeHolder> for ReadableStream
}
/// <https://streams.spec.whatwg.org/#default-reader-release-lock>
fn ReleaseLock(&self) {
if self.stream.get().is_some() {
// step 2 - Perform ! ReadableStreamDefaultReaderRelease(this).
self.release();
fn ReleaseLock(&self) -> Fallible<()> {
if self.stream.get().is_none() {
// Step 1: If this.[[stream]] is undefined, return.
return Ok(());
}
// step 1 - If this.[[stream]] is undefined, return.
// Step 2: Perform !ReadableStreamDefaultReaderRelease(this).
self.release()
}
/// <https://streams.spec.whatwg.org/#generic-reader-closed>
fn Closed(&self) -> Rc<Promise> {
self.closed_promise.borrow().clone()
self.closed()
}
/// <https://streams.spec.whatwg.org/#generic-reader-cancel>
fn Cancel(&self, _cx: SafeJSContext, reason: SafeHandleValue, can_gc: CanGc) -> Rc<Promise> {
if self.stream.get().is_none() {
// If this.[[stream]] is undefined,
// return a promise rejected with a TypeError exception.
let promise = Promise::new(&self.reflector_.global(), can_gc);
promise.reject_error(Error::Type("stream is undefined".to_owned()));
promise
} else {
// Return ! ReadableStreamReaderGenericCancel(this, reason).
self.generic_cancel(reason, can_gc)
}
self.cancel(&self.global(), reason, can_gc)
}
}
impl ReadableStreamGenericReader for ReadableStreamDefaultReader {
fn get_closed_promise(&self) -> Rc<Promise> {
self.closed_promise.borrow().clone()
}
fn set_closed_promise(&self, promise: Rc<Promise>) {
*self.closed_promise.borrow_mut() = promise;
}
fn set_stream(&self, stream: Option<&ReadableStream>) {
self.stream.set(stream);
}
fn get_stream(&self) -> Option<DomRoot<ReadableStream>> {
self.stream.get()
}
fn as_default_reader(&self) -> Option<&ReadableStreamDefaultReader> {
Some(self)
}
}

View file

@ -0,0 +1,164 @@
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
use std::rc::Rc;
use js::jsval::UndefinedValue;
use js::rust::HandleValue as SafeHandleValue;
use super::readablestream::ReaderType;
use super::types::ReadableStream;
use crate::dom::bindings::error::Error;
use crate::dom::bindings::import::module::Fallible;
use crate::dom::bindings::reflector::DomObject;
use crate::dom::bindings::root::{DomRoot, MutNullableDom};
use crate::dom::globalscope::GlobalScope;
use crate::dom::promise::Promise;
use crate::dom::readablestreambyobreader::ReadableStreamBYOBReader;
use crate::dom::readablestreamdefaultreader::ReadableStreamDefaultReader;
use crate::script_runtime::CanGc;
/// <https://streams.spec.whatwg.org/#readablestreamgenericreader>
pub(crate) trait ReadableStreamGenericReader {
/// <https://streams.spec.whatwg.org/#readable-stream-reader-generic-initialize>
#[allow(crown::unrooted_must_root)]
fn generic_initialize(
&self,
global: &GlobalScope,
stream: &ReadableStream,
can_gc: CanGc,
) -> Fallible<()> {
// Set reader.[[stream]] to stream.
self.set_stream(Some(stream));
// Set stream.[[reader]] to reader.
let reader_type = if let Some(default_reader) = self.as_default_reader() {
ReaderType::Default(MutNullableDom::new(Some(default_reader)))
} else if let Some(byob_reader) = self.as_byob_reader() {
ReaderType::BYOB(MutNullableDom::new(Some(byob_reader)))
} else {
unreachable!("Reader must be either Default or BYOB.");
};
stream.set_reader(Some(reader_type));
if stream.is_readable() {
// If stream.[[state]] is "readable
// Set reader.[[closedPromise]] to a new promise.
self.set_closed_promise(Promise::new(global, can_gc));
} else if stream.is_closed() {
// Otherwise, if stream.[[state]] is "closed",
// Set reader.[[closedPromise]] to a promise resolved with undefined.
let cx = GlobalScope::get_cx();
self.set_closed_promise(Promise::new_resolved(global, cx, ())?);
} else {
// Assert: stream.[[state]] is "errored"
assert!(stream.is_errored());
// Set reader.[[closedPromise]] to a promise rejected with stream.[[storedError]].
let cx = GlobalScope::get_cx();
rooted!(in(*cx) let mut error = UndefinedValue());
stream.get_stored_error(error.handle_mut());
self.set_closed_promise(Promise::new_rejected(global, cx, error.handle())?);
// Set reader.[[closedPromise]].[[PromiseIsHandled]] to true
self.get_closed_promise().set_promise_is_handled();
}
Ok(())
}
/// <https://streams.spec.whatwg.org/#readable-stream-reader-generic-cancel>
fn generic_cancel(&self, reason: SafeHandleValue, can_gc: CanGc) -> Rc<Promise> {
// Let stream be reader.[[stream]].
let stream = self.get_stream();
// Assert: stream is not undefined.
let stream =
stream.expect("Reader should have a stream when generic cancel is called into.");
// Return ! ReadableStreamCancel(stream, reason).
stream.cancel(reason, can_gc)
}
/// <https://streams.spec.whatwg.org/#readable-stream-reader-generic-release>
#[allow(unsafe_code)]
fn generic_release(&self) -> Fallible<()> {
// Let stream be reader.[[stream]].
// Assert: stream is not undefined.
assert!(self.get_stream().is_some());
if let Some(stream) = self.get_stream() {
// Assert: stream.[[reader]] is reader.
assert!(stream.has_default_reader());
if stream.is_readable() {
// If stream.[[state]] is "readable", reject reader.[[closedPromise]] with a TypeError exception.
self.get_closed_promise()
.reject_error(Error::Type("stream state is not readable".to_owned()));
} else {
// Otherwise, set reader.[[closedPromise]] to a promise rejected with a TypeError exception.
let cx = GlobalScope::get_cx();
rooted!(in(*cx) let mut error = UndefinedValue());
unsafe {
Error::Type("Cannot release lock due to stream state.".to_owned()).to_jsval(
*cx,
&stream.global(),
error.handle_mut(),
)
};
self.set_closed_promise(
Promise::new_rejected(&stream.global(), cx, error.handle()).unwrap(),
);
}
// Set reader.[[closedPromise]].[[PromiseIsHandled]] to true.
self.get_closed_promise().set_promise_is_handled();
// Perform ! stream.[[controller]].[[ReleaseSteps]]().
stream.perform_release_steps()?;
// Set stream.[[reader]] to undefined.
stream.set_reader(None);
// Set reader.[[stream]] to undefined.
self.set_stream(None);
}
Ok(())
}
/// <https://streams.spec.whatwg.org/#generic-reader-closed>
fn closed(&self) -> Rc<Promise> {
self.get_closed_promise()
}
// <https://streams.spec.whatwg.org/#generic-reader-cancel>
fn cancel(&self, global: &GlobalScope, reason: SafeHandleValue, can_gc: CanGc) -> Rc<Promise> {
if self.get_stream().is_none() {
// If this.[[stream]] is undefined,
// return a promise rejected with a TypeError exception.
let promise = Promise::new(global, can_gc);
promise.reject_error(Error::Type("stream is undefined".to_owned()));
promise
} else {
// Return ! ReadableStreamReaderGenericCancel(this, reason).
self.generic_cancel(reason, can_gc)
}
}
fn set_stream(&self, stream: Option<&ReadableStream>);
fn get_stream(&self) -> Option<DomRoot<ReadableStream>>;
fn set_closed_promise(&self, promise: Rc<Promise>);
fn get_closed_promise(&self) -> Rc<Promise>;
fn as_default_reader(&self) -> Option<&ReadableStreamDefaultReader> {
None
}
fn as_byob_reader(&self) -> Option<&ReadableStreamBYOBReader> {
None
}
}

View file

@ -22,6 +22,7 @@ interface ReadableStreamDefaultReader {
[NewObject]
Promise<ReadableStreamReadResult> read();
[Throws]
undefined releaseLock();
};
ReadableStreamDefaultReader includes ReadableStreamGenericReader;