Demonstrate using dictionary as callback this object.

Signed-off-by: gterzian <2792687+gterzian@users.noreply.github.com>

components/script/Cargo.toml

@@ -63,7 +63,7 @@ image = { workspace = true }
 indexmap = { workspace = true }
 ipc-channel = { workspace = true }
 itertools = { workspace = true }
-js = { package = "mozjs", git = "https://github.com/servo/mozjs", features = ["streams"] }
+js = { package = "mozjs", git = "https://github.com/servo/mozjs" }
 jstraceable_derive = { path = "../jstraceable_derive" }
 keyboard-types = { workspace = true }
 lazy_static = { workspace = true }

components/script/dom/bindings/callback.rs

@@ -15,7 +15,7 @@ use js::jsapi::{
 };
 use js::jsval::{JSVal, ObjectValue, UndefinedValue};
 use js::rust::wrappers::{JS_GetProperty, JS_WrapObject};
-use js::rust::{MutableHandleObject, Runtime};
+use js::rust::{HandleObject, MutableHandleObject, Runtime};
 
 use crate::dom::bindings::codegen::Bindings::WindowBinding::Window_Binding::WindowMethods;
 use crate::dom::bindings::error::{report_pending_exception, Error, Fallible};
@@ -206,9 +206,25 @@ impl CallbackInterface {
     }
 }
 
+pub trait ThisReflector {
+    fn jsobject(&self) -> *mut JSObject;
+}
+
+impl <T: DomObject> ThisReflector for T {
+    fn jsobject(&self) -> *mut JSObject {
+        self.reflector().get_jsobject().get()
+    }
+}
+
+impl <'a> ThisReflector for HandleObject<'a> {
+    fn jsobject(&self) -> *mut JSObject {
+        self.get()
+    }    
+}
+
 /// Wraps the reflector for `p` into the realm of `cx`.
-pub fn wrap_call_this_object<T: DomObject>(cx: JSContext, p: &T, mut rval: MutableHandleObject) {
-    rval.set(p.reflector().get_jsobject().get());
+pub fn wrap_call_this_object<T: ThisReflector>(cx: JSContext, p: &T, mut rval: MutableHandleObject) {
+    rval.set(p.jsobject());
     assert!(!rval.get().is_null());
 
     unsafe {

components/script/dom/bindings/codegen/CodegenRust.py

@@ -949,7 +949,7 @@ def getJSToNativeConversionInfo(type, descriptorProvider, failureCode=None,
 
         if failureCode is None:
             unwrapFailureCode = '''throw_type_error(*cx, "This object is not \
-                    an instance of ReadableStream.");\n'''
+                    an instance of ReadableStream.");\nreturn false;\n'''
         else:
             unwrapFailureCode = failureCode
 
@@ -957,6 +957,7 @@ def getJSToNativeConversionInfo(type, descriptorProvider, failureCode=None,
             """
             {
                 use crate::realms::{AlreadyInRealm, InRealm};
+                use crate::dom::readablestream::ReadableStream;
                 let in_realm_proof = AlreadyInRealm::assert_for_cx(cx);
                 match ReadableStream::from_js(cx, $${val}.get().to_object(), InRealm::Already(&in_realm_proof)) {
                     Ok(val) => val,
@@ -973,7 +974,7 @@ def getJSToNativeConversionInfo(type, descriptorProvider, failureCode=None,
         templateBody = wrapObjectTemplate(templateBody, "None",
                                           isDefinitelyObject, type, failureCode)
 
-        declType = CGGeneric("DomRoot<ReadableStream>")
+        declType = CGGeneric("DomRoot<dom::readablestream::ReadableStream>")
 
         return handleOptional(templateBody, declType,
                               handleDefault("None"))
@@ -7211,7 +7212,7 @@ class CGCallback(CGClass):
               })
         return [ClassMethod(method.name + '_', method.returnType, args,
                             bodyInHeader=True,
-                            templateArgs=["T: DomObject"],
+                            templateArgs=["T: ThisReflector"],
                             body=bodyWithThis,
                             visibility='pub'),
                 ClassMethod(method.name + '__', method.returnType, argsWithoutThis,

components/script/dom/bindings/function.rs

@@ -0,0 +1,48 @@
+/* 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/. */
+
+/// Defines a macro `native_fn!` to create a JavaScript function from a Rust function pointer.
+/// # Example
+/// ```
+/// let js_function: Rc<Function> = native_fn!(my_rust_function, c"myFunction", 2, 0);
+/// ```
+#[macro_export]
+macro_rules! native_fn {
+    ($call:expr, $name:expr, $nargs:expr, $flags:expr) => {{
+        let cx = crate::dom::types::GlobalScope::get_cx();
+        let fun_obj = crate::native_raw_obj_fn!(cx, $call, $name, $nargs, $flags);
+        #[allow(unsafe_code)]
+        unsafe {
+            Function::new(cx, fun_obj)
+        }
+    }};
+}
+
+/// Defines a macro `native_raw_obj_fn!` to create a raw JavaScript function object.
+/// # Example
+/// ```
+/// let raw_function_obj: *mut JSObject = native_raw_obj_fn!(cx, my_rust_function, c"myFunction", 2, 0);
+/// ```
+#[macro_export]
+macro_rules! native_raw_obj_fn {
+    ($cx:expr, $call:expr, $name:expr, $nargs:expr, $flags:expr) => {{
+        #[allow(unsafe_code)]
+        unsafe extern "C" fn wrapper(cx: *mut JSContext, argc: u32, vp: *mut JSVal) -> bool {
+            $call(cx, argc, vp)
+        }
+        #[allow(unsafe_code)]
+        unsafe {
+            let name: &std::ffi::CStr = $name;
+            let raw_fun = crate::dom::bindings::import::module::jsapi::JS_NewFunction(
+                *$cx,
+                Some(wrapper),
+                $nargs,
+                $flags,
+                name.as_ptr() as *const std::ffi::c_char,
+            );
+            assert!(!raw_fun.is_null());
+            crate::dom::bindings::import::module::jsapi::JS_GetFunctionObject(raw_fun)
+        }
+    }};
+}

components/script/dom/bindings/import.rs

@@ -19,7 +19,7 @@ pub mod base {
 
     pub use crate::dom::bindings::callback::{
         wrap_call_this_object, CallSetup, CallbackContainer, CallbackFunction, CallbackInterface,
-        CallbackObject, ExceptionHandling,
+        CallbackObject, ExceptionHandling, ThisReflector,
     };
     pub use crate::dom::bindings::codegen::Bindings::AudioNodeBinding::{
         ChannelCountMode, ChannelCountModeValues, ChannelInterpretation,

components/script/dom/bindings/interface.rs

@@ -146,7 +146,6 @@ pub unsafe fn create_global_object(
     let mut options = RealmOptions::default();
     options.creationOptions_.traceGlobal_ = Some(trace);
     options.creationOptions_.sharedMemoryAndAtomics_ = false;
-    options.creationOptions_.streams_ = true;
     select_compartment(cx, &mut options);
 
     let principal = ServoJSPrincipals::new(origin);

components/script/dom/bindings/mod.rs

@@ -141,6 +141,7 @@ pub mod constant;
 pub mod conversions;
 pub mod error;
 pub mod finalize;
+pub mod function;
 pub mod guard;
 pub mod htmlconstructor;
 pub mod import;

components/script/dom/bytelengthqueuingstrategy.rs

@@ -0,0 +1,98 @@
+/* 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 dom_struct::dom_struct;
+use js::gc::{HandleValue, MutableHandleValue};
+use js::jsapi::{CallArgs, JSContext};
+use js::jsval::JSVal;
+use js::rust::HandleObject;
+
+use super::bindings::codegen::Bindings::FunctionBinding::Function;
+use super::bindings::codegen::Bindings::QueuingStrategyBinding::{
+    ByteLengthQueuingStrategyMethods, QueuingStrategyInit,
+};
+use super::bindings::import::module::{DomObject, DomRoot, Fallible, Reflector};
+use super::bindings::reflector::reflect_dom_object_with_proto;
+use super::types::GlobalScope;
+use crate::dom::bindings::import::module::get_dictionary_property;
+use crate::native_fn;
+
+#[dom_struct]
+pub struct ByteLengthQueuingStrategy {
+    reflector_: Reflector,
+    high_water_mark: f64,
+}
+
+#[allow(non_snake_case)]
+impl ByteLengthQueuingStrategy {
+    /// <https://streams.spec.whatwg.org/#blqs-constructor>
+    pub fn Constructor(
+        global: &GlobalScope,
+        proto: Option<HandleObject>,
+        init: &QueuingStrategyInit,
+    ) -> DomRoot<Self> {
+        Self::new(global, proto, init.highWaterMark)
+    }
+
+    pub fn new_inherited(init: f64) -> Self {
+        Self {
+            reflector_: Reflector::new(),
+            high_water_mark: init,
+        }
+    }
+
+    pub fn new(global: &GlobalScope, proto: Option<HandleObject>, init: f64) -> DomRoot<Self> {
+        reflect_dom_object_with_proto(Box::new(Self::new_inherited(init)), global, proto)
+    }
+}
+
+impl ByteLengthQueuingStrategyMethods for ByteLengthQueuingStrategy {
+    /// <https://streams.spec.whatwg.org/#blqs-high-water-mark>
+    fn HighWaterMark(&self) -> f64 {
+        self.high_water_mark
+    }
+
+    /// <https://streams.spec.whatwg.org/#blqs-size>
+    fn GetSize(&self) -> Fallible<Rc<Function>> {
+        let global = self.reflector_.global();
+        // Return this's relevant global object's byte length queuing strategy
+        // size function.
+        if let Some(fun) = global.get_byte_length_queuing_strategy_size() {
+            return Ok(fun);
+        }
+
+        // Step 1. Let steps be the following steps, given chunk
+        // Note: See ByteLengthQueuingStrategySize instead.
+
+        // Step 2. Let F be !CreateBuiltinFunction(steps, 1, "size", « »,
+        // globalObject’s relevant Realm).
+        let fun = native_fn!(byte_length_queuing_strategy_size, c"size", 1, 0);
+        // Step 3. Set globalObject’s byte length queuing strategy size function to
+        // a Function that represents a reference to F,
+        // with callback context equal to globalObject’s relevant settings object.
+        global.set_byte_length_queuing_strategy_size(fun.clone());
+        Ok(fun)
+    }
+}
+
+/// <https://streams.spec.whatwg.org/#byte-length-queuing-strategy-size-function>
+#[allow(unsafe_code)]
+pub unsafe fn byte_length_queuing_strategy_size(
+    cx: *mut JSContext,
+    argc: u32,
+    vp: *mut JSVal,
+) -> bool {
+    let args = CallArgs::from_vp(vp, argc);
+    // Step 1.1: Return ? GetV(chunk, "byteLength").
+    rooted!(in(cx) let object = HandleValue::from_raw(args.get(0)).to_object());
+    get_dictionary_property(
+        cx,
+        object.handle(),
+        "byteLength",
+        MutableHandleValue::from_raw(args.rval()),
+    )
+    .unwrap_or(false)
+}

components/script/dom/countqueuingstrategy.rs

@@ -0,0 +1,122 @@
+/* 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 dom_struct::dom_struct;
+use js::jsapi::{CallArgs, JSContext};
+use js::jsval::{Int32Value, JSVal};
+use js::rust::HandleObject;
+
+use super::bindings::codegen::Bindings::FunctionBinding::Function;
+use super::bindings::codegen::Bindings::QueuingStrategyBinding::{
+    CountQueuingStrategyMethods, QueuingStrategy, QueuingStrategyInit, QueuingStrategySize,
+};
+use super::bindings::import::module::{DomObject, DomRoot, Error, Fallible, Reflector};
+use super::bindings::reflector::reflect_dom_object_with_proto;
+use super::bytelengthqueuingstrategy::byte_length_queuing_strategy_size;
+use super::types::GlobalScope;
+use crate::{native_fn, native_raw_obj_fn};
+
+#[dom_struct]
+pub struct CountQueuingStrategy {
+    reflector_: Reflector,
+    high_water_mark: f64,
+}
+
+#[allow(non_snake_case)]
+impl CountQueuingStrategy {
+    /// <https://streams.spec.whatwg.org/#cqs-constructor>
+    pub fn Constructor(
+        global: &GlobalScope,
+        proto: Option<HandleObject>,
+        init: &QueuingStrategyInit,
+    ) -> DomRoot<Self> {
+        Self::new(global, proto, init.highWaterMark)
+    }
+
+    pub fn new_inherited(init: f64) -> Self {
+        Self {
+            reflector_: Reflector::new(),
+            high_water_mark: init,
+        }
+    }
+
+    pub fn new(global: &GlobalScope, proto: Option<HandleObject>, init: f64) -> DomRoot<Self> {
+        reflect_dom_object_with_proto(Box::new(Self::new_inherited(init)), global, proto)
+    }
+}
+
+impl CountQueuingStrategyMethods for CountQueuingStrategy {
+    /// <https://streams.spec.whatwg.org/#cqs-high-water-mark>
+    fn HighWaterMark(&self) -> f64 {
+        self.high_water_mark
+    }
+
+    /// <https://streams.spec.whatwg.org/#cqs-size>
+    fn GetSize(&self) -> Fallible<Rc<Function>> {
+        let global = self.reflector_.global();
+        // Return this's relevant global object's count queuing strategy
+        // size function.
+        if let Some(fun) = global.get_count_queuing_strategy_size() {
+            return Ok(fun);
+        }
+
+        // Step 1. Let steps be the following steps, given chunk
+        // Note: See ByteLengthQueuingStrategySize instead.
+
+        // Step 2. Let F be !CreateBuiltinFunction(steps, 1, "size", « »,
+        // globalObject’s relevant Realm).
+        let fun = native_fn!(byte_length_queuing_strategy_size, c"size", 0, 0);
+        // Step 3. Set globalObject’s count queuing strategy size function to
+        // a Function that represents a reference to F,
+        // with callback context equal to globalObject’s relevant settings object.
+        global.set_count_queuing_strategy_size(fun.clone());
+        Ok(fun)
+    }
+}
+
+/// <https://streams.spec.whatwg.org/#count-queuing-strategy-size-function>
+#[allow(unsafe_code)]
+pub unsafe fn count_queuing_strategy_size(_cx: *mut JSContext, argc: u32, vp: *mut JSVal) -> bool {
+    let args = CallArgs::from_vp(vp, argc);
+    // Step 1.1. Return 1.
+    args.rval().set(Int32Value(1));
+    true
+}
+
+/// Extract the high water mark from a QueuingStrategy.
+/// If the high water mark is not set, return the default value.
+///
+/// <https://streams.spec.whatwg.org/#validate-and-normalize-high-water-mark>
+pub fn extract_high_water_mark(strategy: &QueuingStrategy, default_hwm: f64) -> Result<f64, Error> {
+    if strategy.highWaterMark.is_none() {
+        return Ok(default_hwm);
+    }
+
+    let high_water_mark = strategy.highWaterMark.unwrap();
+    if high_water_mark.is_nan() || high_water_mark < 0.0 {
+        return Err(Error::Range(
+            "High water mark must be a non-negative number.".to_string(),
+        ));
+    }
+
+    Ok(high_water_mark)
+}
+
+/// Extract the size algorithm from a QueuingStrategy.
+/// If the size algorithm is not set, return a fallback function which always returns 1.
+///
+/// <https://streams.spec.whatwg.org/#make-size-algorithm-from-size-function>
+pub fn extract_size_algorithm(strategy: &QueuingStrategy) -> Rc<QueuingStrategySize> {
+    if strategy.size.is_none() {
+        let cx = GlobalScope::get_cx();
+        let fun_obj = native_raw_obj_fn!(cx, count_queuing_strategy_size, c"size", 0, 0);
+        #[allow(unsafe_code)]
+        unsafe {
+            QueuingStrategySize::new(cx, fun_obj).clone()
+        };
+    }
+    strategy.size.as_ref().unwrap().clone()
+}

components/script/dom/globalscope.rs

@@ -3,7 +3,7 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 use std::borrow::Cow;
-use std::cell::Cell;
+use std::cell::{Cell, OnceCell};
 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, VecDeque};
 use std::ops::Index;
@@ -63,6 +63,7 @@ use super::bindings::trace::HashMapTracedValues;
 use crate::dom::bindings::cell::{DomRefCell, RefMut};
 use crate::dom::bindings::codegen::Bindings::BroadcastChannelBinding::BroadcastChannelMethods;
 use crate::dom::bindings::codegen::Bindings::EventSourceBinding::EventSource_Binding::EventSourceMethods;
+use crate::dom::bindings::codegen::Bindings::FunctionBinding::Function;
 use crate::dom::bindings::codegen::Bindings::ImageBitmapBinding::{
     ImageBitmapOptions, ImageBitmapSource,
 };
@@ -81,6 +82,7 @@ use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
 use crate::dom::bindings::settings_stack::{entry_global, incumbent_global, AutoEntryScript};
 use crate::dom::bindings::str::DOMString;
 use crate::dom::bindings::structuredclone;
+use crate::dom::bindings::trace::CustomTraceable;
 use crate::dom::bindings::utils::to_frozen_array;
 use crate::dom::bindings::weakref::{DOMTracker, WeakRef};
 use crate::dom::blob::Blob;
@@ -106,9 +108,10 @@ use crate::dom::paintworkletglobalscope::PaintWorkletGlobalScope;
 use crate::dom::performance::Performance;
 use crate::dom::performanceobserver::VALID_ENTRY_TYPES;
 use crate::dom::promise::Promise;
-use crate::dom::readablestream::{ExternalUnderlyingSource, ReadableStream};
+use crate::dom::readablestream::ReadableStream;
 use crate::dom::serviceworker::ServiceWorker;
 use crate::dom::serviceworkerregistration::ServiceWorkerRegistration;
+use crate::dom::underlyingsourcecontainer::UnderlyingSourceType;
 use crate::dom::window::Window;
 use crate::dom::workerglobalscope::WorkerGlobalScope;
 use crate::dom::workletglobalscope::WorkletGlobalScope;
@@ -345,6 +348,20 @@ pub struct GlobalScope {
 
     /// Is considered in a secure context
     inherited_secure_context: Option<bool>,
+
+    /// The byte length queuing strategy size function that will be initialized once
+    /// `size` getter of `ByteLengthQueuingStrategy` is called.
+    ///
+    /// <https://streams.spec.whatwg.org/#byte-length-queuing-strategy-size-function>
+    #[ignore_malloc_size_of = "Rc<T> is hard"]
+    byte_length_queuing_strategy_size_function: OnceCell<Rc<Function>>,
+
+    /// The count queuing strategy size function that will be initialized once
+    /// `size` getter of `CountQueuingStrategy` is called.
+    ///
+    /// <https://streams.spec.whatwg.org/#count-queuing-strategy-size-function>
+    #[ignore_malloc_size_of = "Rc<T> is hard"]
+    count_queuing_strategy_size_function: OnceCell<Rc<Function>>,
 }
 
 /// A wrapper for glue-code between the ipc router and the event-loop.
@@ -805,6 +822,8 @@ impl GlobalScope {
             console_count_map: Default::default(),
             dynamic_modules: DomRefCell::new(DynamicModuleList::new()),
             inherited_secure_context,
+            byte_length_queuing_strategy_size_function: OnceCell::new(),
+            count_queuing_strategy_size_function: OnceCell::new(),
         }
     }
 
@@ -2005,7 +2024,7 @@ impl GlobalScope {
 
         let stream = ReadableStream::new_with_external_underlying_source(
             self,
-            ExternalUnderlyingSource::Blob(size),
+            UnderlyingSourceType::Blob(size),
         );
 
         let recv = self.send_msg(file_id);
@@ -3275,6 +3294,33 @@ impl GlobalScope {
     pub(crate) fn dynamic_module_list(&self) -> RefMut<DynamicModuleList> {
         self.dynamic_modules.borrow_mut()
     }
+
+    pub(crate) fn set_byte_length_queuing_strategy_size(&self, function: Rc<Function>) {
+        if let Err(_) = self
+            .byte_length_queuing_strategy_size_function
+            .set(function)
+        {
+            warn!("byte length queuing strategy size function is set twice.");
+        };
+    }
+
+    pub(crate) fn get_byte_length_queuing_strategy_size(&self) -> Option<Rc<Function>> {
+        self.byte_length_queuing_strategy_size_function
+            .get()
+            .map(|s| s.clone())
+    }
+
+    pub(crate) fn set_count_queuing_strategy_size(&self, function: Rc<Function>) {
+        if let Err(_) = self.count_queuing_strategy_size_function.set(function) {
+            warn!("count queuing strategy size function is set twice.");
+        };
+    }
+
+    pub(crate) fn get_count_queuing_strategy_size(&self) -> Option<Rc<Function>> {
+        self.count_queuing_strategy_size_function
+            .get()
+            .map(|s| s.clone())
+    }
 }
 
 /// Returns the Rust global scope from a JS global object.

components/script/dom/mod.rs

@@ -242,6 +242,7 @@ pub mod bluetoothremotegattserver;
 pub mod bluetoothremotegattservice;
 pub mod bluetoothuuid;
 pub mod broadcastchannel;
+pub mod bytelengthqueuingstrategy;
 pub mod canvasgradient;
 pub mod canvaspattern;
 pub mod canvasrenderingcontext2d;
@@ -255,6 +256,7 @@ pub mod comment;
 pub mod compositionevent;
 pub mod console;
 pub mod constantsourcenode;
+pub mod countqueuingstrategy;
 mod create;
 pub mod crypto;
 pub mod css;
@@ -505,7 +507,12 @@ pub mod promiserejectionevent;
 pub mod radionodelist;
 pub mod range;
 pub mod raredata;
+pub mod readablebytestreamcontroller;
 pub mod readablestream;
+pub mod readablestreambyobreader;
+pub mod readablestreambyobrequest;
+pub mod readablestreamdefaultcontroller;
+pub mod readablestreamdefaultreader;
 pub mod request;
 pub mod resizeobserver;
 pub mod resizeobserverentry;
@@ -568,6 +575,7 @@ pub mod trackevent;
 pub mod transitionevent;
 pub mod treewalker;
 pub mod uievent;
+pub mod underlyingsourcecontainer;
 pub mod url;
 pub mod urlhelper;
 pub mod urlsearchparams;

components/script/dom/readablebytestreamcontroller.rs

@@ -0,0 +1,56 @@
+/* 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 dom_struct::dom_struct;
+use js::rust::HandleValue as SafeHandleValue;
+
+use crate::dom::bindings::codegen::Bindings::ReadableByteStreamControllerBinding::ReadableByteStreamControllerMethods;
+use crate::dom::bindings::import::module::{Error, Fallible};
+use crate::dom::bindings::reflector::Reflector;
+use crate::dom::bindings::root::DomRoot;
+use crate::script_runtime::JSContext as SafeJSContext;
+
+/// <https://streams.spec.whatwg.org/#readablebytestreamcontroller>
+#[dom_struct]
+pub struct ReadableByteStreamController {
+    reflector_: Reflector,
+}
+
+impl ReadableByteStreamControllerMethods for ReadableByteStreamController {
+    /// <https://streams.spec.whatwg.org/#rbs-controller-byob-request>
+    fn GetByobRequest(
+        &self,
+    ) -> Fallible<Option<DomRoot<super::readablestreambyobrequest::ReadableStreamBYOBRequest>>>
+    {
+        // TODO
+        Err(Error::NotFound)
+    }
+
+    /// <https://streams.spec.whatwg.org/#rbs-controller-desired-size>
+    fn GetDesiredSize(&self) -> Option<f64> {
+        // TODO
+        None
+    }
+
+    /// <https://streams.spec.whatwg.org/#rbs-controller-close>
+    fn Close(&self) -> Fallible<()> {
+        // TODO
+        Err(Error::NotFound)
+    }
+
+    /// <https://streams.spec.whatwg.org/#rbs-controller-enqueue>
+    fn Enqueue(
+        &self,
+        _chunk: js::gc::CustomAutoRooterGuard<js::typedarray::ArrayBufferView>,
+    ) -> Fallible<()> {
+        // TODO
+        Err(Error::NotFound)
+    }
+
+    /// <https://streams.spec.whatwg.org/#rbs-controller-error>
+    fn Error(&self, _cx: SafeJSContext, _e: SafeHandleValue) -> Fallible<()> {
+        // TODO
+        Err(Error::NotFound)
+    }
+}

components/script/dom/readablestream.rs

@@ -2,109 +2,183 @@
  * 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::cell::{Cell, RefCell};
-use std::os::raw::c_void;
+use std::cell::Cell;
 use std::ptr::{self, NonNull};
 use std::rc::Rc;
-use std::slice;
 
 use dom_struct::dom_struct;
-use js::glue::{
-    CreateReadableStreamUnderlyingSource, DeleteReadableStreamUnderlyingSource,
-    ReadableStreamUnderlyingSourceTraps,
-};
-use js::jsapi::{
-    AutoRequireNoGC, HandleObject, HandleValue, Heap, IsReadableStream, JSContext, JSObject,
-    JS_GetArrayBufferViewData, NewReadableExternalSourceStreamObject, ReadableStreamClose,
-    ReadableStreamDefaultReaderRead, ReadableStreamError, ReadableStreamGetReader,
-    ReadableStreamIsDisturbed, ReadableStreamIsLocked, ReadableStreamIsReadable,
-    ReadableStreamReaderMode, ReadableStreamReaderReleaseLock, ReadableStreamUnderlyingSource,
-    ReadableStreamUpdateDataAvailableFromSource, UnwrapReadableStream,
-};
-use js::jsval::{JSVal, UndefinedValue};
-use js::rust::{HandleValue as SafeHandleValue, IntoHandle};
+use js::jsapi::JSObject;
+use js::jsval::{ObjectValue, UndefinedValue};
+use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
 
+use crate::dom::bindings::cell::DomRefCell;
+use crate::dom::bindings::codegen::Bindings::QueuingStrategyBinding::QueuingStrategy;
+use crate::dom::bindings::codegen::Bindings::ReadableStreamBinding::{
+    ReadableStreamGetReaderOptions, ReadableStreamMethods,
+};
+use crate::dom::bindings::codegen::Bindings::ReadableStreamDefaultReaderBinding::ReadableStreamDefaultReaderMethods;
+use crate::dom::bindings::codegen::Bindings::UnderlyingSourceBinding::UnderlyingSource as JsUnderlyingSource;
 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::{
+    ReadableStreamDefaultControllerOrReadableByteStreamController as Controller,
+    ReadableStreamDefaultReaderOrReadableStreamBYOBReader as ReadableStreamReader,
+};
 use crate::dom::bindings::reflector::{reflect_dom_object, DomObject, Reflector};
-use crate::dom::bindings::root::DomRoot;
-use crate::dom::bindings::settings_stack::{AutoEntryScript, AutoIncumbentScript};
+use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
 use crate::dom::bindings::utils::get_dictionary_property;
 use crate::dom::globalscope::GlobalScope;
 use crate::dom::promise::Promise;
+use crate::dom::readablebytestreamcontroller::ReadableByteStreamController;
+use crate::dom::readablestreambyobreader::ReadableStreamBYOBReader;
+use crate::dom::readablestreamdefaultcontroller::ReadableStreamDefaultController;
+use crate::dom::readablestreamdefaultreader::{ReadRequest, ReadableStreamDefaultReader};
+use crate::dom::underlyingsourcecontainer::UnderlyingSourceType;
 use crate::js::conversions::FromJSValConvertible;
-use crate::realms::{enter_realm, InRealm};
+use crate::realms::InRealm;
 use crate::script_runtime::JSContext as SafeJSContext;
 
-static UNDERLYING_SOURCE_TRAPS: ReadableStreamUnderlyingSourceTraps =
-    ReadableStreamUnderlyingSourceTraps {
-        requestData: Some(request_data),
-        writeIntoReadRequestBuffer: Some(write_into_read_request_buffer),
-        cancel: Some(cancel),
-        onClosed: Some(close),
-        onErrored: Some(error),
-        finalize: Some(finalize),
-    };
+/// <https://streams.spec.whatwg.org/#readablestream-state>
+#[derive(Default, JSTraceable, MallocSizeOf)]
+pub enum ReadableStreamState {
+    #[default]
+    Readable,
+    Closed,
+    Errored,
+}
 
+/// <https://streams.spec.whatwg.org/#readablestream-controller>
+#[derive(JSTraceable, MallocSizeOf)]
+#[crown::unrooted_must_root_lint::must_root]
+pub enum ControllerType {
+    /// <https://streams.spec.whatwg.org/#readablebytestreamcontroller>
+    Byte(Dom<ReadableByteStreamController>),
+    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller>
+    Default(Dom<ReadableStreamDefaultController>),
+}
+
+/// <https://streams.spec.whatwg.org/#readablestream-readerr>
+#[derive(JSTraceable, MallocSizeOf)]
+#[crown::unrooted_must_root_lint::must_root]
+pub enum ReaderType {
+    /// <https://streams.spec.whatwg.org/#readablestreambyobreader>
+    BYOB(MutNullableDom<ReadableStreamBYOBReader>),
+    /// <https://streams.spec.whatwg.org/#readablestreamdefaultreader>
+    Default(MutNullableDom<ReadableStreamDefaultReader>),
+}
+
+/// <https://streams.spec.whatwg.org/#rs-class>
 #[dom_struct]
 pub struct ReadableStream {
     reflector_: Reflector,
-    #[ignore_malloc_size_of = "SM handles JS values"]
-    js_stream: Heap<*mut JSObject>,
-    #[ignore_malloc_size_of = "SM handles JS values"]
-    js_reader: Heap<*mut JSObject>,
-    has_reader: Cell<bool>,
-    #[ignore_malloc_size_of = "Rc is hard"]
-    external_underlying_source: Option<Rc<ExternalUnderlyingSourceController>>,
+
+    /// <https://streams.spec.whatwg.org/#readablestream-controller>
+    controller: ControllerType,
+
+    /// <https://streams.spec.whatwg.org/#readablestream-storederror>
+    stored_error: DomRefCell<Option<Error>>,
+
+    /// <https://streams.spec.whatwg.org/#readablestream-disturbed>
+    disturbed: Cell<bool>,
+
+    /// <https://streams.spec.whatwg.org/#readablestream-reader>
+    reader: ReaderType,
+
+    /// <https://streams.spec.whatwg.org/#readablestream-state>
+    state: DomRefCell<ReadableStreamState>,
 }
 
 impl ReadableStream {
-    fn new_inherited(
-        external_underlying_source: Option<Rc<ExternalUnderlyingSourceController>>,
-    ) -> ReadableStream {
+    /// <https://streams.spec.whatwg.org/#rs-constructor>
+    #[allow(non_snake_case)]
+    pub fn Constructor(
+        cx: SafeJSContext,
+        global: &GlobalScope,
+        _proto: Option<SafeHandleObject>,
+        underlying_source: Option<*mut JSObject>,
+        _strategy: &QueuingStrategy,
+    ) -> Fallible<DomRoot<Self>> {
+        // Step 1
+        rooted!(in(*cx) let underlying_source_obj = underlying_source.unwrap_or(ptr::null_mut()));
+        // Step 2
+        let underlying_source_dict = if !underlying_source_obj.is_null() {
+            rooted!(in(*cx) let obj_val = ObjectValue(underlying_source_obj.get()));
+            match JsUnderlyingSource::new(cx, obj_val.handle()) {
+                Ok(ConversionResult::Success(val)) => val,
+                Ok(ConversionResult::Failure(error)) => return Err(Error::Type(error.to_string())),
+                _ => {
+                    return Err(Error::Type(
+                        "Unknown format for underlying source.".to_string(),
+                    ))
+                },
+            }
+        } else {
+            JsUnderlyingSource::empty()
+        };
+
+        let controller = if underlying_source_dict.type_.is_some() {
+            // TODO: byte controller.
+            todo!()
+        } else {
+            ReadableStreamDefaultController::new(
+                global,
+                UnderlyingSourceType::Js(underlying_source_dict),
+            )
+        };
+        Ok(ReadableStream::new(
+            global,
+            Controller::ReadableStreamDefaultController(controller),
+        ))
+    }
+
+    #[allow(crown::unrooted_must_root)]
+    fn new_inherited(controller: Controller) -> ReadableStream {
+        let reader = match &controller {
+            Controller::ReadableStreamDefaultController(_) => {
+                ReaderType::Default(MutNullableDom::new(None))
+            },
+            Controller::ReadableByteStreamController(_) => {
+                ReaderType::BYOB(MutNullableDom::new(None))
+            },
+        };
         ReadableStream {
             reflector_: Reflector::new(),
-            js_stream: Heap::default(),
-            js_reader: Heap::default(),
-            has_reader: Default::default(),
-            external_underlying_source,
+            controller: match controller {
+                Controller::ReadableStreamDefaultController(root) => {
+                    ControllerType::Default(Dom::from_ref(&*root))
+                },
+                Controller::ReadableByteStreamController(root) => {
+                    ControllerType::Byte(Dom::from_ref(&*root))
+                },
+            },
+            stored_error: DomRefCell::new(None),
+            disturbed: Default::default(),
+            reader: reader,
+            state: DomRefCell::new(ReadableStreamState::Readable),
         }
     }
 
-    fn new(
-        global: &GlobalScope,
-        external_underlying_source: Option<Rc<ExternalUnderlyingSourceController>>,
-    ) -> DomRoot<ReadableStream> {
-        reflect_dom_object(
-            Box::new(ReadableStream::new_inherited(external_underlying_source)),
-            global,
-        )
+    fn new(global: &GlobalScope, controller: Controller) -> DomRoot<ReadableStream> {
+        reflect_dom_object(Box::new(ReadableStream::new_inherited(controller)), global)
     }
 
     /// Used from RustCodegen.py
+    /// TODO: remove here and its use in codegen.
     #[allow(unsafe_code)]
     pub unsafe fn from_js(
-        cx: SafeJSContext,
-        obj: *mut JSObject,
-        realm: InRealm,
+        _cx: SafeJSContext,
+        _obj: *mut JSObject,
+        _realm: InRealm,
     ) -> Result<DomRoot<ReadableStream>, ()> {
-        if !IsReadableStream(obj) {
-            return Err(());
-        }
-
-        let global = GlobalScope::from_safe_context(cx, realm);
-
-        let stream = ReadableStream::new(&global, None);
-        stream.js_stream.set(UnwrapReadableStream(obj));
-
-        Ok(stream)
+        Err(())
     }
 
     /// Build a stream backed by a Rust source that has already been read into memory.
     pub fn new_from_bytes(global: &GlobalScope, bytes: Vec<u8>) -> DomRoot<ReadableStream> {
         let stream = ReadableStream::new_with_external_underlying_source(
             global,
-            ExternalUnderlyingSource::Memory(bytes.len()),
+            UnderlyingSourceType::Memory(bytes.len()),
         );
         stream.enqueue_native(bytes);
         stream.close_native();
@@ -112,400 +186,254 @@ impl ReadableStream {
     }
 
     /// Build a stream backed by a Rust underlying source.
+    /// Note: external sources are always paired with a default controller.
     #[allow(unsafe_code)]
     pub fn new_with_external_underlying_source(
         global: &GlobalScope,
-        source: ExternalUnderlyingSource,
+        source: UnderlyingSourceType,
     ) -> DomRoot<ReadableStream> {
-        let _ar = enter_realm(global);
-        let _ais = AutoIncumbentScript::new(global);
-        let cx = GlobalScope::get_cx();
-
-        let source = Rc::new(ExternalUnderlyingSourceController::new(source));
-
-        let stream = ReadableStream::new(global, Some(source.clone()));
-
-        unsafe {
-            let js_wrapper = CreateReadableStreamUnderlyingSource(
-                &UNDERLYING_SOURCE_TRAPS,
-                &*source as *const _ as *const c_void,
-            );
-
-            rooted!(in(*cx)
-                let js_stream = NewReadableExternalSourceStreamObject(
-                    *cx,
-                    js_wrapper,
-                    ptr::null_mut(),
-                    HandleObject::null(),
-                )
-            );
-
-            stream.js_stream.set(UnwrapReadableStream(js_stream.get()));
-        }
+        assert!(source.is_native());
+        let controller = ReadableStreamDefaultController::new(global, source);
+        let stream = ReadableStream::new(
+            global,
+            Controller::ReadableStreamDefaultController(controller.clone()),
+        );
+        controller.set_stream(&stream);
 
         stream
     }
 
+    /// Call into the pull steps of the controller,
+    /// as part of
+    /// <https://streams.spec.whatwg.org/#readable-stream-default-reader-read>
+    pub fn perform_pull_steps(&self, read_request: ReadRequest) {
+        match self.controller {
+            ControllerType::Default(ref controller) => controller.perform_pull_steps(read_request),
+            _ => todo!(),
+        }
+    }
+
+    /// <https://streams.spec.whatwg.org/#readable-stream-add-read-request>
+    pub fn add_read_request(&self, read_request: ReadRequest) {
+        match self.reader {
+            ReaderType::Default(ref reader) => {
+                let Some(reader) = reader.get() else {
+                    panic!("Attempt to read stream chunk without having acquired a reader.");
+                };
+                reader.add_read_request(read_request);
+            },
+            _ => unreachable!("Adding a read request can only be done on a default reader."),
+        }
+    }
+
     /// Get a pointer to the underlying JS object.
+    /// TODO: remove,
+    /// by using at call point the `ReadableStream` directly instead of a JSObject.
     pub fn get_js_stream(&self) -> NonNull<JSObject> {
-        NonNull::new(self.js_stream.get())
+        NonNull::new(*self.reflector().get_jsobject())
             .expect("Couldn't get a non-null pointer to JS stream object.")
     }
 
-    #[allow(unsafe_code)]
+    /// Endpoint to enqueue chunks directly from Rust.
+    /// Note: in other use cases this call happens via the controller.
     pub fn enqueue_native(&self, bytes: Vec<u8>) {
-        let global = self.global();
-        let _ar = enter_realm(&*global);
-        let cx = GlobalScope::get_cx();
-
-        let handle = unsafe { self.js_stream.handle() };
-
-        self.external_underlying_source
-            .as_ref()
-            .expect("No external source to enqueue bytes.")
-            .enqueue_chunk(cx, handle, bytes);
-    }
-
-    #[allow(unsafe_code)]
-    pub fn error_native(&self, error: Error) {
-        let global = self.global();
-        let _ar = enter_realm(&*global);
-        let cx = GlobalScope::get_cx();
-
-        unsafe {
-            rooted!(in(*cx) let mut js_error = UndefinedValue());
-            error.to_jsval(*cx, &global, js_error.handle_mut());
-            ReadableStreamError(
-                *cx,
-                self.js_stream.handle(),
-                js_error.handle().into_handle(),
-            );
+        match self.controller {
+            ControllerType::Default(ref controller) => controller.enqueue_native(bytes),
+            _ => unreachable!(
+                "Enqueueing chunk to a stream from Rust on other than default controller"
+            ),
         }
     }
 
-    #[allow(unsafe_code)]
-    pub fn close_native(&self) {
-        let global = self.global();
-        let _ar = enter_realm(&*global);
-        let cx = GlobalScope::get_cx();
-
-        let handle = unsafe { self.js_stream.handle() };
-
-        self.external_underlying_source
-            .as_ref()
-            .expect("No external source to close.")
-            .close(cx, handle);
+    /// <https://streams.spec.whatwg.org/#readable-stream-error>
+    /// Note: in other use cases this call happens via the controller.
+    pub fn error_native(&self, _error: Error) {
+        *self.state.borrow_mut() = ReadableStreamState::Errored;
+        match self.controller {
+            ControllerType::Default(ref controller) => controller.error(),
+            _ => unreachable!("Native closing a stream with a non-default controller"),
+        }
     }
 
-    /// Does the stream have all data in memory?
+    /// <https://streams.spec.whatwg.org/#readable-stream-close>
+    /// Note: in other use cases this call happens via the controller.
+    pub fn close_native(&self) {
+        match self.controller {
+            ControllerType::Default(ref controller) => controller.close(),
+            _ => unreachable!("Native closing a stream with a non-default controller"),
+        }
+    }
+
+    /// Returns a boolean reflecting whether the stream has all data in memory.
+    /// Useful for native source integration only.
     pub fn in_memory(&self) -> bool {
-        self.external_underlying_source
-            .as_ref()
-            .map(|source| source.in_memory())
-            .unwrap_or(false)
+        match self.controller {
+            ControllerType::Default(ref controller) => controller.in_memory(),
+            _ => unreachable!(
+                "Checking if source is in memory for a stream with a non-default controller"
+            ),
+        }
     }
 
     /// Return bytes for synchronous use, if the stream has all data in memory.
+    /// Useful for native source integration only.
     pub fn get_in_memory_bytes(&self) -> Option<Vec<u8>> {
-        self.external_underlying_source
-            .as_ref()
-            .and_then(|source| source.get_in_memory_bytes())
+        match self.controller {
+            ControllerType::Default(ref controller) => controller.get_in_memory_bytes(),
+            _ => unreachable!("Getting in-memory bytes for a stream with a non-default controller"),
+        }
     }
 
-    /// Acquires a reader and locks the stream,
-    /// must be done before `read_a_chunk`.
-    #[allow(unsafe_code)]
+    /// Native call to
+    /// <https://streams.spec.whatwg.org/#acquire-readable-stream-reader>
+    /// TODO: restructure this on related methods so the caller gets a reader?
     pub fn start_reading(&self) -> Result<(), ()> {
-        if self.is_locked() || self.is_disturbed() {
+        if self.is_locked() {
             return Err(());
         }
-
         let global = self.global();
-        let _ar = enter_realm(&*global);
-        let cx = GlobalScope::get_cx();
-
-        unsafe {
-            rooted!(in(*cx) let reader = ReadableStreamGetReader(
-                *cx,
-                self.js_stream.handle(),
-                ReadableStreamReaderMode::Default,
-            ));
-
-            // Note: the stream is locked to the reader.
-            self.js_reader.set(reader.get());
+        match self.reader {
+            ReaderType::Default(ref reader) => {
+                reader.set(Some(&*ReadableStreamDefaultReader::new(&*global, self)))
+            },
+            _ => unreachable!("Native start reading can only be done on a default reader."),
         }
-
-        self.has_reader.set(true);
         Ok(())
     }
 
-    /// Read a chunk from the stream,
-    /// must be called after `start_reading`,
-    /// and before `stop_reading`.
-    #[allow(unsafe_code)]
+    /// Native call to
+    /// <https://streams.spec.whatwg.org/#readable-stream-default-reader-read>
+    /// TODO: restructure this on related methods so the caller reads from a reader?
     pub fn read_a_chunk(&self) -> Rc<Promise> {
-        if !self.has_reader.get() {
-            panic!("Attempt to read stream chunk without having acquired a reader.");
-        }
-
-        let global = self.global();
-        let _ar = enter_realm(&*global);
-        let _aes = AutoEntryScript::new(&global);
-
-        let cx = GlobalScope::get_cx();
-
-        unsafe {
-            rooted!(in(*cx) let promise_obj = ReadableStreamDefaultReaderRead(
-                *cx,
-                self.js_reader.handle(),
-            ));
-            Promise::new_with_js_promise(promise_obj.handle(), cx)
+        match self.reader {
+            ReaderType::Default(ref reader) => {
+                let Some(reader) = reader.get() else {
+                    panic!("Attempt to read stream chunk without having acquired a reader.");
+                };
+                reader.Read()
+            },
+            _ => unreachable!("Native reading a chunk can only be done on a default reader."),
         }
     }
 
-    /// Releases the lock on the reader,
-    /// must be done after `start_reading`.
-    #[allow(unsafe_code)]
+    /// Native call to
+    /// <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaultreaderrelease>
+    /// TODO: restructure this on related methods so the caller releases a reader?
     pub fn stop_reading(&self) {
-        if !self.has_reader.get() {
-            panic!("ReadableStream::stop_reading called on a readerless stream.");
-        }
-
-        self.has_reader.set(false);
-
-        let global = self.global();
-        let _ar = enter_realm(&*global);
-        let cx = GlobalScope::get_cx();
-
-        unsafe {
-            ReadableStreamReaderReleaseLock(*cx, self.js_reader.handle());
-            // Note: is this the way to nullify the Heap?
-            self.js_reader.set(ptr::null_mut());
+        match self.reader {
+            ReaderType::Default(ref reader) => {
+                let Some(rooted_reader) = reader.get() else {
+                    panic!("Attempt to read stream chunk without having acquired a reader.");
+                };
+                rooted_reader.ReleaseLock();
+                reader.set(None);
+            },
+            _ => unreachable!("Native stop reading a chunk can only be done on a default reader."),
         }
     }
 
-    #[allow(unsafe_code)]
     pub fn is_locked(&self) -> bool {
-        // If we natively took a reader, we're locked.
-        if self.has_reader.get() {
-            return true;
+        match self.reader {
+            ReaderType::Default(ref reader) => reader.get().is_some(),
+            ReaderType::BYOB(ref reader) => reader.get().is_some(),
         }
-
-        // Otherwise, still double-check that script didn't lock the stream.
-        let cx = GlobalScope::get_cx();
-        let mut locked_or_disturbed = false;
-
-        unsafe {
-            ReadableStreamIsLocked(*cx, self.js_stream.handle(), &mut locked_or_disturbed);
-        }
-
-        locked_or_disturbed
     }
 
-    #[allow(unsafe_code)]
     pub fn is_disturbed(&self) -> bool {
-        // Check that script didn't disturb the stream.
-        let cx = GlobalScope::get_cx();
-        let mut locked_or_disturbed = false;
-
-        unsafe {
-            ReadableStreamIsDisturbed(*cx, self.js_stream.handle(), &mut locked_or_disturbed);
-        }
-
-        locked_or_disturbed
+        self.disturbed.get()
     }
-}
 
-#[allow(unsafe_code)]
-unsafe extern "C" fn request_data(
-    source: *const c_void,
-    cx: *mut JSContext,
-    stream: HandleObject,
-    desired_size: usize,
-) {
-    let source = &*(source as *const ExternalUnderlyingSourceController);
-    source.pull(SafeJSContext::from_ptr(cx), stream, desired_size);
-}
+    pub fn is_closed(&self) -> bool {
+        matches!(*self.state.borrow(), ReadableStreamState::Closed)
+    }
 
-#[allow(unsafe_code)]
-unsafe extern "C" fn write_into_read_request_buffer(
-    source: *const c_void,
-    _cx: *mut JSContext,
-    _stream: HandleObject,
-    chunk: HandleObject,
-    length: usize,
-    bytes_written: *mut usize,
-) {
-    let source = &*(source as *const ExternalUnderlyingSourceController);
-    let mut is_shared_memory = false;
-    let buffer = JS_GetArrayBufferViewData(
-        *chunk,
-        &mut is_shared_memory,
-        &AutoRequireNoGC { _address: 0 },
-    );
-    assert!(!is_shared_memory);
-    let slice = slice::from_raw_parts_mut(buffer as *mut u8, length);
-    source.write_into_buffer(slice);
+    pub fn is_errored(&self) -> bool {
+        matches!(*self.state.borrow(), ReadableStreamState::Errored)
+    }
 
-    // Currently we're always able to completely fulfill the write request.
-    *bytes_written = length;
-}
+    pub fn is_readable(&self) -> bool {
+        matches!(*self.state.borrow(), ReadableStreamState::Readable)
+    }
 
-#[allow(unsafe_code)]
-unsafe extern "C" fn cancel(
-    _source: *const c_void,
-    _cx: *mut JSContext,
-    _stream: HandleObject,
-    _reason: HandleValue,
-    _resolve_to: *mut JSVal,
-) {
-}
-
-#[allow(unsafe_code)]
-unsafe extern "C" fn close(_source: *const c_void, _cx: *mut JSContext, _stream: HandleObject) {}
-
-#[allow(unsafe_code)]
-unsafe extern "C" fn error(
-    _source: *const c_void,
-    _cx: *mut JSContext,
-    _stream: HandleObject,
-    _reason: HandleValue,
-) {
-}
-
-#[allow(unsafe_code)]
-unsafe extern "C" fn finalize(source: *mut ReadableStreamUnderlyingSource) {
-    DeleteReadableStreamUnderlyingSource(source);
-}
-
-pub enum ExternalUnderlyingSource {
-    /// Facilitate partial integration with sources
-    /// that are currently read into memory.
-    Memory(usize),
-    /// A blob as underlying source, with a known total size.
-    Blob(usize),
-    /// A fetch response as underlying source.
-    FetchResponse,
-}
-
-#[derive(JSTraceable, MallocSizeOf)]
-struct ExternalUnderlyingSourceController {
-    /// Loosely matches the underlying queue,
-    /// <https://streams.spec.whatwg.org/#internal-queues>
-    buffer: RefCell<Vec<u8>>,
-    /// Has the stream been closed by native code?
-    closed: Cell<bool>,
-    /// Does this stream contains all it's data in memory?
-    in_memory: Cell<bool>,
-}
-
-impl ExternalUnderlyingSourceController {
-    fn new(source: ExternalUnderlyingSource) -> ExternalUnderlyingSourceController {
-        let (buffer, in_mem) = match source {
-            ExternalUnderlyingSource::Blob(size) => (Vec::with_capacity(size), false),
-            ExternalUnderlyingSource::Memory(size) => (Vec::with_capacity(size), true),
-            ExternalUnderlyingSource::FetchResponse => (vec![], false),
-        };
-        ExternalUnderlyingSourceController {
-            buffer: RefCell::new(buffer),
-            closed: Cell::new(false),
-            in_memory: Cell::new(in_mem),
+    pub fn has_default_reader(&self) -> bool {
+        match self.reader {
+            ReaderType::Default(ref reader) => reader.get().is_some(),
+            ReaderType::BYOB(_) => false,
         }
     }
 
-    /// Does the stream have all data in memory?
-    pub fn in_memory(&self) -> bool {
-        self.in_memory.get()
-    }
-
-    /// Return bytes synchronously if the stream has all data in memory.
-    pub fn get_in_memory_bytes(&self) -> Option<Vec<u8>> {
-        if self.in_memory.get() {
-            return Some(self.buffer.borrow().clone());
+    /// <https://streams.spec.whatwg.org/#readable-stream-get-num-read-requests>
+    pub fn get_num_read_requests(&self) -> usize {
+        assert!(self.has_default_reader());
+        match self.reader {
+            ReaderType::Default(ref reader) => {
+                let reader = reader
+                    .get()
+                    .expect("Stream must have a reader when get num read requests is called into.");
+                reader.get_num_read_requests()
+            },
+            _ => unreachable!(
+                "Stream must have a default reader when get num read requests is called into."
+            ),
         }
-        None
     }
 
-    /// Signal available bytes if the stream is currently readable.
-    #[allow(unsafe_code)]
-    fn maybe_signal_available_bytes(
+    /// <https://streams.spec.whatwg.org/#readable-stream-fulfill-read-request>
+    pub fn fulfill_read_request(&self, chunk: Vec<u8>, done: bool) {
+        assert!(self.has_default_reader());
+        match self.reader {
+            ReaderType::Default(ref reader) => {
+                let reader = reader
+                    .get()
+                    .expect("Stream must have a reader when a read request is fulfilled.");
+                let request = reader.remove_read_request();
+                if !done {
+                    request.chunk_steps(chunk);
+                } else {
+                    request.close_steps();
+                }
+            },
+            _ => unreachable!(
+                "Stream must have a default reader when fulfill read requests is called into."
+            ),
+        }
+    }
+}
+
+impl ReadableStreamMethods for ReadableStream {
+    /// <https://streams.spec.whatwg.org/#rs-locked>
+    fn Locked(&self) -> bool {
+        // TODO
+        false
+    }
+
+    /// <https://streams.spec.whatwg.org/#rs-cancel>
+    fn Cancel(&self, _cx: SafeJSContext, _reason: SafeHandleValue) -> Rc<Promise> {
+        // TODO
+        Promise::new(&self.reflector_.global())
+    }
+
+    /// <https://streams.spec.whatwg.org/#rs-get-reader>
+    fn GetReader(
         &self,
-        cx: SafeJSContext,
-        stream: HandleObject,
-        available: usize,
-    ) {
-        if available == 0 {
-            return;
+        _options: &ReadableStreamGetReaderOptions,
+    ) -> Fallible<ReadableStreamReader> {
+        if self.is_locked() {
+            return Err(Error::Type("Stream is locked".to_string()));
         }
-        unsafe {
-            let mut readable = false;
-            if !ReadableStreamIsReadable(*cx, stream, &mut readable) {
-                return;
-            }
-            if readable {
-                ReadableStreamUpdateDataAvailableFromSource(*cx, stream, available as u32);
-            }
+        match self.reader {
+            ReaderType::Default(ref reader) => {
+                reader.set(Some(&*ReadableStreamDefaultReader::new(
+                    &*self.global(),
+                    self,
+                )));
+                return Ok(ReadableStreamReader::ReadableStreamDefaultReader(
+                    reader.get().unwrap(),
+                ));
+            },
+            _ => todo!(),
         }
     }
-
-    /// Close a currently readable js stream.
-    #[allow(unsafe_code)]
-    fn maybe_close_js_stream(&self, cx: SafeJSContext, stream: HandleObject) {
-        unsafe {
-            let mut readable = false;
-            if !ReadableStreamIsReadable(*cx, stream, &mut readable) {
-                return;
-            }
-            if readable {
-                ReadableStreamClose(*cx, stream);
-            }
-        }
-    }
-
-    fn close(&self, cx: SafeJSContext, stream: HandleObject) {
-        self.closed.set(true);
-        self.maybe_close_js_stream(cx, stream);
-    }
-
-    fn enqueue_chunk(&self, cx: SafeJSContext, stream: HandleObject, mut chunk: Vec<u8>) {
-        let available = {
-            let mut buffer = self.buffer.borrow_mut();
-            chunk.append(&mut buffer);
-            *buffer = chunk;
-            buffer.len()
-        };
-        self.maybe_signal_available_bytes(cx, stream, available);
-    }
-
-    #[allow(unsafe_code)]
-    fn pull(&self, cx: SafeJSContext, stream: HandleObject, _desired_size: usize) {
-        // Note: for pull sources,
-        // this would be the time to ask for a chunk.
-
-        if self.closed.get() {
-            return self.maybe_close_js_stream(cx, stream);
-        }
-
-        let available = {
-            let buffer = self.buffer.borrow();
-            buffer.len()
-        };
-
-        self.maybe_signal_available_bytes(cx, stream, available);
-    }
-
-    fn get_chunk_with_length(&self, length: usize) -> Vec<u8> {
-        let mut buffer = self.buffer.borrow_mut();
-        let buffer_len = buffer.len();
-        assert!(buffer_len >= length);
-        buffer.split_off(buffer_len - length)
-    }
-
-    fn write_into_buffer(&self, dest: &mut [u8]) {
-        let length = dest.len();
-        let chunk = self.get_chunk_with_length(length);
-        dest.copy_from_slice(chunk.as_slice());
-    }
 }
 
 #[allow(unsafe_code)]

components/script/dom/readablestreambyobreader.rs

@@ -0,0 +1,75 @@
+/* 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 dom_struct::dom_struct;
+use js::gc::CustomAutoRooterGuard;
+use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
+use js::typedarray::ArrayBufferView;
+
+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::globalscope::GlobalScope;
+use crate::dom::promise::Promise;
+use crate::dom::readablestream::ReadableStream;
+use crate::script_runtime::JSContext as SafeJSContext;
+
+/// <https://streams.spec.whatwg.org/#readablestreambyobreader>
+#[dom_struct]
+pub struct ReadableStreamBYOBReader {
+    reflector_: Reflector,
+}
+
+impl ReadableStreamBYOBReader {
+    /// <https://streams.spec.whatwg.org/#byob-reader-constructor>
+    #[allow(non_snake_case)]
+    pub fn Constructor(
+        _global: &GlobalScope,
+        _proto: Option<SafeHandleObject>,
+        _stream: DomRoot<ReadableStream>,
+    ) -> Fallible<DomRoot<Self>> {
+        // TODO
+        Err(Error::NotFound)
+    }
+
+    fn new_inherited() -> ReadableStreamBYOBReader {
+        ReadableStreamBYOBReader {
+            reflector_: Reflector::new(),
+        }
+    }
+
+    fn new(global: &GlobalScope) -> DomRoot<ReadableStreamBYOBReader> {
+        reflect_dom_object(Box::new(ReadableStreamBYOBReader::new_inherited()), global)
+    }
+}
+
+impl ReadableStreamBYOBReaderMethods for ReadableStreamBYOBReader {
+    /// <https://streams.spec.whatwg.org/#byob-reader-read>
+    fn Read(&self, _view: CustomAutoRooterGuard<ArrayBufferView>) -> Rc<Promise> {
+        // TODO
+        Promise::new(&self.reflector_.global())
+    }
+
+    /// <https://streams.spec.whatwg.org/#byob-reader-release-lock>
+    fn ReleaseLock(&self) -> Fallible<()> {
+        // TODO
+        Err(Error::NotFound)
+    }
+
+    /// <https://streams.spec.whatwg.org/#generic-reader-closed>
+    fn Closed(&self) -> Rc<Promise> {
+        // TODO
+        Promise::new(&self.reflector_.global())
+    }
+
+    /// <https://streams.spec.whatwg.org/#generic-reader-cancel>
+    fn Cancel(&self, _cx: SafeJSContext, _reason: SafeHandleValue) -> Rc<Promise> {
+        // TODO
+        Promise::new(&self.reflector_.global())
+    }
+}

components/script/dom/readablestreambyobrequest.rs

@@ -0,0 +1,39 @@
+/* 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 dom_struct::dom_struct;
+
+use crate::dom::bindings::codegen::Bindings::ReadableStreamBYOBRequestBinding::ReadableStreamBYOBRequestMethods;
+use crate::dom::bindings::import::module::{Error, Fallible};
+use crate::dom::bindings::reflector::Reflector;
+use crate::script_runtime::JSContext as SafeJSContext;
+
+/// <https://streams.spec.whatwg.org/#readablestreambyobrequest>
+#[dom_struct]
+pub struct ReadableStreamBYOBRequest {
+    reflector_: Reflector,
+}
+
+impl ReadableStreamBYOBRequestMethods for ReadableStreamBYOBRequest {
+    /// <https://streams.spec.whatwg.org/#rs-byob-request-view>
+    fn GetView(&self, _cx: SafeJSContext) -> Option<js::typedarray::ArrayBufferView> {
+        // TODO
+        None
+    }
+
+    /// <https://streams.spec.whatwg.org/#rs-byob-request-respond>
+    fn Respond(&self, _bytes_written: u64) -> Fallible<()> {
+        // TODO
+        Err(Error::NotFound)
+    }
+
+    /// <https://streams.spec.whatwg.org/#rs-byob-request-respond-with-new-view>
+    fn RespondWithNewView(
+        &self,
+        _view: js::gc::CustomAutoRooterGuard<js::typedarray::ArrayBufferView>,
+    ) -> Fallible<()> {
+        // TODO
+        Err(Error::NotFound)
+    }
+}

components/script/dom/readablestreamdefaultcontroller.rs

@@ -0,0 +1,288 @@
+/* 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::cell::RefCell;
+use std::collections::VecDeque;
+use std::rc::Rc;
+
+use dom_struct::dom_struct;
+use js::jsapi::Heap;
+use js::rust::{HandleValue as SafeHandleValue, HandleValue};
+
+use crate::dom::bindings::codegen::Bindings::ReadableStreamDefaultControllerBinding::{
+    ReadableStreamDefaultControllerMethods, ValueWithSize,
+};
+use crate::dom::bindings::import::module::UnionTypes::ReadableStreamDefaultControllerOrReadableByteStreamController as Controller;
+use crate::dom::bindings::import::module::{Error, Fallible};
+use crate::dom::bindings::refcounted::Trusted;
+use crate::dom::bindings::reflector::{reflect_dom_object, DomObject, Reflector};
+use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
+use crate::dom::globalscope::GlobalScope;
+use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
+use crate::dom::readablestream::ReadableStream;
+use crate::dom::readablestreamdefaultreader::ReadRequest;
+use crate::dom::underlyingsourcecontainer::{UnderlyingSourceContainer, UnderlyingSourceType};
+use crate::realms::{enter_realm, InRealm};
+use crate::script_runtime::{JSContext, JSContext as SafeJSContext};
+
+/// The fulfillment handler for
+/// <https://streams.spec.whatwg.org/#readable-stream-default-controller-call-pull-if-needed>
+#[derive(Clone, JSTraceable, MallocSizeOf)]
+#[allow(crown::unrooted_must_root)]
+struct PullAlgorithmFulfillmentHandler {
+    controller: Dom<ReadableStreamDefaultController>,
+}
+
+impl Callback for PullAlgorithmFulfillmentHandler {
+    /// Handle fufillment of pull algo promise.
+    fn callback(&self, _cx: JSContext, _v: HandleValue, _realm: InRealm) {
+        todo!();
+    }
+}
+
+/// The rejection handler for
+/// <https://streams.spec.whatwg.org/#readable-stream-default-controller-call-pull-if-needed>
+#[derive(Clone, JSTraceable, MallocSizeOf)]
+#[allow(crown::unrooted_must_root)]
+struct PullAlgorithmRejectionHandler {
+    controller: Dom<ReadableStreamDefaultController>,
+}
+
+impl Callback for PullAlgorithmRejectionHandler {
+    /// Handle rejection of pull algo promise.
+    fn callback(&self, _cx: JSContext, _v: HandleValue, _realm: InRealm) {
+        todo!();
+    }
+}
+
+/// <https://streams.spec.whatwg.org/#value-with-size>
+#[derive(JSTraceable)]
+#[allow(crown::unrooted_must_root)]
+pub enum EnqueuedValue {
+    /// A value enqueued from Rust.
+    Native(Vec<u8>),
+    /// A Js value.
+    Js(ValueWithSize),
+}
+
+/// <https://streams.spec.whatwg.org/#queue-with-sizes>
+#[derive(Default, JSTraceable, MallocSizeOf)]
+pub struct QueueWithSizes {
+    total_size: usize,
+    #[ignore_malloc_size_of = "Rc is hard"]
+    queue: VecDeque<EnqueuedValue>,
+}
+
+impl QueueWithSizes {
+    /// <https://streams.spec.whatwg.org/#dequeue-value>
+    fn dequeue_value(&mut self) -> EnqueuedValue {
+        self.queue
+            .pop_front()
+            .expect("Buffer cannot be empty when dequeue value is called into.")
+    }
+
+    /// <https://streams.spec.whatwg.org/#enqueue-value-with-size>
+    fn enqueue_value_with_size(&mut self, value: EnqueuedValue) {
+        self.queue.push_back(value);
+    }
+
+    fn is_empty(&self) -> bool {
+        self.queue.is_empty()
+    }
+
+    /// Only used with native sources.
+    fn get_in_memory_bytes(&self) -> Vec<u8> {
+        self.queue
+            .iter()
+            .flat_map(|value| {
+                let EnqueuedValue::Native(chunk) = value else {
+                    unreachable!(
+                        "`get_in_memory_bytes` can only be called on a queue with native values."
+                    )
+                };
+                chunk.clone()
+            })
+            .collect()
+    }
+}
+
+/// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller>
+#[dom_struct]
+pub struct ReadableStreamDefaultController {
+    reflector_: Reflector,
+
+    /// <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-queue>
+    queue: RefCell<QueueWithSizes>,
+
+    underlying_source: Dom<UnderlyingSourceContainer>,
+
+    stream: MutNullableDom<ReadableStream>,
+}
+
+impl ReadableStreamDefaultController {
+    fn new_inherited(
+        global: &GlobalScope,
+        underlying_source_type: UnderlyingSourceType,
+    ) -> ReadableStreamDefaultController {
+        ReadableStreamDefaultController {
+            reflector_: Reflector::new(),
+            queue: RefCell::new(Default::default()),
+            stream: MutNullableDom::new(None),
+            underlying_source: Dom::from_ref(&*UnderlyingSourceContainer::new(
+                global,
+                underlying_source_type,
+            )),
+        }
+    }
+    pub fn new(
+        global: &GlobalScope,
+        underlying_source: UnderlyingSourceType,
+    ) -> DomRoot<ReadableStreamDefaultController> {
+        reflect_dom_object(
+            Box::new(ReadableStreamDefaultController::new_inherited(
+                global,
+                underlying_source,
+            )),
+            global,
+        )
+    }
+
+    pub fn set_stream(&self, stream: &ReadableStream) {
+        self.stream.set(Some(stream))
+    }
+
+    /// <https://streams.spec.whatwg.org/#dequeue-value>
+    fn dequeue_value(&self) -> EnqueuedValue {
+        let mut queue = self.queue.borrow_mut();
+        queue.dequeue_value()
+    }
+
+    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-should-call-pull>
+    fn should_pull(&self) -> bool {
+        // TODO: implement the algo.
+        true
+    }
+
+    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-call-pull-if-needed>
+    fn call_pull_if_needed(&self) {
+        if !self.should_pull() {
+            return;
+        }
+
+        let global = self.global();
+        let controller = Controller::ReadableStreamDefaultController(DomRoot::from_ref(self));
+        if let Some(promise) = self.underlying_source.call_pull_algorithm(controller) {
+            let fulfillment_handler = Box::new(PullAlgorithmFulfillmentHandler {
+                controller: Dom::from_ref(self),
+            });
+            let rejection_handler = Box::new(PullAlgorithmRejectionHandler {
+                controller: Dom::from_ref(self),
+            });
+            let handler = PromiseNativeHandler::new(
+                &global,
+                Some(fulfillment_handler),
+                Some(rejection_handler),
+            );
+
+            let realm = enter_realm(&*global);
+            let comp = InRealm::Entered(&realm);
+            promise.append_native_handler(&handler, comp);
+        }
+    }
+
+    /// <https://streams.spec.whatwg.org/#ref-for-abstract-opdef-readablestreamcontroller-pullsteps>
+    pub fn perform_pull_steps(&self, read_request: ReadRequest) {
+        // if queue contains bytes, perform chunk steps.
+        if !self.queue.borrow().is_empty() {
+            // TODO: use <https://streams.spec.whatwg.org/#readablestreamdefaultcontroller-strategysizealgorithm>
+            let chunk = self.dequeue_value();
+
+            // TODO: handle close requested.
+
+            self.call_pull_if_needed();
+
+            if let EnqueuedValue::Native(chunk) = chunk {
+                read_request.chunk_steps(chunk);
+            }
+        }
+
+        // else, append read request to reader.
+        self.stream
+            .get()
+            .expect("Controller must have a stream when pull steps are called into.")
+            .add_read_request(read_request);
+
+        self.call_pull_if_needed();
+    }
+
+    /// Native call to
+    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-enqueue>
+    pub fn enqueue_native(&self, chunk: Vec<u8>) {
+        let stream = self
+            .stream
+            .get()
+            .expect("Controller must have a stream when a chunk is enqueued.");
+        if stream.is_locked() && stream.get_num_read_requests() > 0 {
+            stream.fulfill_read_request(chunk, false);
+            return;
+        }
+
+        // TODO: strategy size algo.
+
+        // <https://streams.spec.whatwg.org/#enqueue-value-with-size>
+        let mut queue = self.queue.borrow_mut();
+        queue.enqueue_value_with_size(EnqueuedValue::Native(chunk));
+
+        self.call_pull_if_needed();
+    }
+
+    /// Does the stream have all data in memory?
+    pub fn in_memory(&self) -> bool {
+        self.underlying_source.in_memory()
+    }
+
+    /// Return bytes synchronously if the stream has all data in memory.
+    pub fn get_in_memory_bytes(&self) -> Option<Vec<u8>> {
+        if self.underlying_source.in_memory() {
+            return Some(self.queue.borrow().get_in_memory_bytes());
+        }
+        None
+    }
+
+    /// <https://streams.spec.whatwg.org/#readable-stream-default-controller-close>
+    pub fn close(&self) {
+        todo!()
+    }
+
+    /// <https://streams.spec.whatwg.org/#ref-for-readable-stream-error>
+    pub fn error(&self) {
+        todo!()
+    }
+}
+
+impl ReadableStreamDefaultControllerMethods for ReadableStreamDefaultController {
+    /// <https://streams.spec.whatwg.org/#rs-default-controller-desired-size>
+    fn GetDesiredSize(&self) -> Option<f64> {
+        // TODO
+        None
+    }
+
+    /// <https://streams.spec.whatwg.org/#rs-default-controller-close>
+    fn Close(&self) -> Fallible<()> {
+        // TODO
+        Err(Error::NotFound)
+    }
+
+    /// <https://streams.spec.whatwg.org/#rs-default-controller-enqueue>
+    fn Enqueue(&self, _cx: SafeJSContext, _chunk: SafeHandleValue) -> Fallible<()> {
+        // TODO
+        Err(Error::NotFound)
+    }
+
+    /// <https://streams.spec.whatwg.org/#rs-default-controller-error>
+    fn Error(&self, _cx: SafeJSContext, _e: SafeHandleValue) -> Fallible<()> {
+        // TODO
+        Err(Error::NotFound)
+    }
+}

components/script/dom/readablestreamdefaultreader.rs

@@ -0,0 +1,212 @@
+/* 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::collections::VecDeque;
+use std::ptr;
+use std::rc::Rc;
+
+use dom_struct::dom_struct;
+use js::conversions::ToJSValConvertible;
+use js::jsapi::Heap;
+use js::jsval::UndefinedValue;
+use js::rust::{HandleObject as SafeHandleObject, HandleValue as SafeHandleValue};
+
+use crate::dom::bindings::cell::DomRefCell;
+use crate::dom::bindings::codegen::Bindings::ReadableStreamDefaultReaderBinding::{
+    ReadableStreamDefaultReaderMethods, ReadableStreamReadResult,
+};
+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::{Dom, DomRoot};
+use crate::dom::bindings::trace::RootedTraceableBox;
+use crate::dom::bindings::utils::set_dictionary_property;
+use crate::dom::globalscope::GlobalScope;
+use crate::dom::promise::Promise;
+use crate::dom::readablestream::ReadableStream;
+use crate::dom::readablestreamdefaultcontroller::EnqueuedValue;
+use crate::script_runtime::JSContext as SafeJSContext;
+
+/// <https://streams.spec.whatwg.org/#read-request>
+/// For now only one variant: the one matching a `read` call.
+#[derive(JSTraceable)]
+pub enum ReadRequest {
+    /// <https://streams.spec.whatwg.org/#default-reader-read>
+    Read(Rc<Promise>),
+}
+
+impl ReadRequest {
+    /// <https://streams.spec.whatwg.org/#read-request-chunk-steps>
+    #[allow(unsafe_code)]
+    #[allow(crown::unrooted_must_root)]
+    pub fn chunk_steps(&self, chunk: Vec<u8>) {
+        match self {
+            ReadRequest::Read(promise) => {
+                let cx = GlobalScope::get_cx();
+                rooted!(in(*cx) let mut rval = UndefinedValue());
+                let result = RootedTraceableBox::new(Heap::default());
+                unsafe {
+                    chunk.to_jsval(*cx, rval.handle_mut());
+                    result.set(*rval);
+                }
+                let result = ReadableStreamReadResult {
+                    done: Some(false),
+                    value: result,
+                };
+                promise.resolve_native(&result);
+            },
+        }
+    }
+
+    /// <https://streams.spec.whatwg.org/#ref-for-read-request-close-step>
+    #[allow(crown::unrooted_must_root)]
+    pub fn close_steps(&self) {
+        match self {
+            ReadRequest::Read(promise) => {
+                let cx = GlobalScope::get_cx();
+                rooted!(in(*cx) let mut rval = UndefinedValue());
+                let result = RootedTraceableBox::new(Heap::default());
+                result.set(*rval);
+                let result = ReadableStreamReadResult {
+                    done: Some(true),
+                    value: result,
+                };
+                promise.resolve_native(&result);
+            },
+        }
+    }
+
+    /// <https://streams.spec.whatwg.org/#ref-for-read-request-close-step>
+    pub fn error_steps(&self) {
+        match self {
+            // TODO: pass error type.
+            ReadRequest::Read(promise) => promise.reject_native(&()),
+        }
+    }
+}
+
+/// <https://streams.spec.whatwg.org/#readablestreamdefaultreader>
+#[dom_struct]
+pub struct ReadableStreamDefaultReader {
+    reflector_: Reflector,
+
+    /// <https://streams.spec.whatwg.org/#readablestreamgenericreader-stream>
+    /// TODO: use MutNullableDom
+    stream: Dom<ReadableStream>,
+
+    #[ignore_malloc_size_of = "Rc is hard"]
+    read_requests: DomRefCell<VecDeque<ReadRequest>>,
+
+    /// <https://streams.spec.whatwg.org/#readablestreamgenericreader-closedpromise>
+    #[ignore_malloc_size_of = "Rc is hard"]
+    closed_promise: Rc<Promise>,
+}
+
+impl ReadableStreamDefaultReader {
+    /// <https://streams.spec.whatwg.org/#default-reader-constructor>
+    #[allow(non_snake_case)]
+    pub fn Constructor(
+        _global: &GlobalScope,
+        _proto: Option<SafeHandleObject>,
+        _stream: DomRoot<ReadableStream>,
+    ) -> Fallible<DomRoot<Self>> {
+        // TODO
+        Err(Error::NotFound)
+    }
+
+    fn new_inherited(
+        stream: &ReadableStream,
+        closed_promise: Rc<Promise>,
+    ) -> ReadableStreamDefaultReader {
+        ReadableStreamDefaultReader {
+            reflector_: Reflector::new(),
+            stream: Dom::from_ref(stream),
+            read_requests: DomRefCell::new(Default::default()),
+            closed_promise,
+        }
+    }
+
+    /// <https://streams.spec.whatwg.org/#readable-stream-reader-generic-initialize>
+    pub fn new(
+        global: &GlobalScope,
+        stream: &ReadableStream,
+    ) -> DomRoot<ReadableStreamDefaultReader> {
+        let promise = Promise::new(global);
+        if stream.is_closed() {
+            promise.resolve_native(&());
+        }
+        if stream.is_errored() {
+            promise.reject_native(&());
+        }
+        reflect_dom_object(
+            Box::new(ReadableStreamDefaultReader::new_inherited(stream, promise)),
+            global,
+        )
+    }
+
+    /// <https://streams.spec.whatwg.org/#readable-stream-add-read-request>
+    pub fn add_read_request(&self, read_request: ReadRequest) {
+        self.read_requests.borrow_mut().push_back(read_request);
+    }
+
+    /// <https://streams.spec.whatwg.org/#readable-stream-get-num-read-requests>
+    pub fn get_num_read_requests(&self) -> usize {
+        self.read_requests.borrow().len()
+    }
+
+    /// <https://streams.spec.whatwg.org/#readable-stream-error>
+    pub fn error(&self, _error: Error) {
+        self.closed_promise.reject_native(&());
+        for request in self.read_requests.borrow_mut().drain(0..) {
+            request.error_steps();
+        }
+    }
+
+    /// The removal steps of <https://streams.spec.whatwg.org/#readable-stream-fulfill-read-request>
+    pub fn remove_read_request(&self) -> ReadRequest {
+        self.read_requests
+            .borrow_mut()
+            .pop_front()
+            .expect("Reader must have read request when remove is called into.")
+    }
+}
+
+impl ReadableStreamDefaultReaderMethods for ReadableStreamDefaultReader {
+    /// <https://streams.spec.whatwg.org/#default-reader-read>
+    fn Read(&self) -> Rc<Promise> {
+        let promise = Promise::new(&self.reflector_.global());
+
+        self.stream
+            .perform_pull_steps(ReadRequest::Read(promise.clone()));
+
+        promise
+    }
+
+    /// <https://streams.spec.whatwg.org/#default-reader-release-lock>
+    fn ReleaseLock(&self) {
+        if self.stream.is_readable() {
+            self.closed_promise.reject_native(&());
+        }
+
+        // TODO: https://streams.spec.whatwg.org/#readable-stream-reader-generic-release
+
+        // TODO: use TypeError.
+        // <https://streams.spec.whatwg.org/#abstract-opdef-readablestreamdefaultreadererrorreadrequests>
+        for request in self.read_requests.borrow_mut().drain(0..) {
+            request.error_steps();
+        }
+    }
+
+    /// <https://streams.spec.whatwg.org/#generic-reader-closed>
+    fn Closed(&self) -> Rc<Promise> {
+        // TODO
+        Promise::new(&self.reflector_.global())
+    }
+
+    /// <https://streams.spec.whatwg.org/#generic-reader-cancel>
+    fn Cancel(&self, _cx: SafeJSContext, _reason: SafeHandleValue) -> Rc<Promise> {
+        // TODO
+        Promise::new(&self.reflector_.global())
+    }
+}

components/script/dom/response.rs

@@ -30,7 +30,8 @@ use crate::dom::bindings::str::{ByteString, USVString};
 use crate::dom::globalscope::GlobalScope;
 use crate::dom::headers::{is_obs_text, is_vchar, Guard, Headers};
 use crate::dom::promise::Promise;
-use crate::dom::readablestream::{ExternalUnderlyingSource, ReadableStream};
+use crate::dom::readablestream::ReadableStream;
+use crate::dom::underlyingsourcecontainer::UnderlyingSourceType;
 use crate::script_runtime::{JSContext as SafeJSContext, StreamConsumer};
 
 #[dom_struct]
@@ -59,7 +60,7 @@ impl Response {
     pub fn new_inherited(global: &GlobalScope) -> Response {
         let stream = ReadableStream::new_with_external_underlying_source(
             global,
-            ExternalUnderlyingSource::FetchResponse,
+            UnderlyingSourceType::FetchResponse,
         );
         Response {
             reflector_: Reflector::new(),

components/script/dom/underlyingsourcecontainer.rs

@@ -0,0 +1,118 @@
+/* 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 dom_struct::dom_struct;
+use js::jsval::UndefinedValue;
+
+use crate::dom::bindings::callback::ExceptionHandling;
+use crate::dom::bindings::codegen::Bindings::UnderlyingSourceBinding::UnderlyingSource as JsUnderlyingSource;
+use crate::dom::bindings::import::module::UnionTypes::ReadableStreamDefaultControllerOrReadableByteStreamController as Controller;
+use crate::dom::bindings::reflector::{reflect_dom_object_with_proto, DomObject, Reflector};
+use crate::dom::bindings::root::DomRoot;
+use crate::dom::globalscope::GlobalScope;
+use crate::dom::promise::Promise;
+use crate::dom::readablestreamdefaultcontroller::ReadableStreamDefaultController;
+use crate::js::conversions::ToJSValConvertible;
+use js::jsapi::JSObject;
+use std::ptr;
+
+/// <https://streams.spec.whatwg.org/#underlying-source-api>
+/// The `Js` variant corresponds to
+/// the JavaScript object representing the underlying source.
+/// The other variants are native sources in Rust.
+#[derive(JSTraceable)]
+pub enum UnderlyingSourceType {
+    /// Facilitate partial integration with sources
+    /// that are currently read into memory.
+    Memory(usize),
+    /// A blob as underlying source, with a known total size.
+    Blob(usize),
+    /// A fetch response as underlying source.
+    FetchResponse,
+    /// A JS object as underlying source.
+    Js(JsUnderlyingSource),
+}
+
+impl UnderlyingSourceType {
+    /// Is the source backed by a Rust native source?
+    pub fn is_native(&self) -> bool {
+        match self {
+            UnderlyingSourceType::Memory(_) |
+            UnderlyingSourceType::Blob(_) |
+            UnderlyingSourceType::FetchResponse => true,
+            _ => false,
+        }
+    }
+
+    /// Does the source have all data in memory?
+    pub fn in_memory(&self) -> bool {
+        matches!(self, UnderlyingSourceType::Memory(_))
+    }
+}
+
+/// Wrapper around the underlying source.
+/// Useful because `Call_` requires the "this object" to impl DomObject.
+#[dom_struct]
+pub struct UnderlyingSourceContainer {
+    reflector_: Reflector,
+    #[ignore_malloc_size_of = "JsUnderlyingSource implemented in SM."]
+    underlying_source_type: UnderlyingSourceType,
+}
+
+impl UnderlyingSourceContainer {
+    fn new_inherited(underlying_source_type: UnderlyingSourceType) -> UnderlyingSourceContainer {
+        UnderlyingSourceContainer {
+            reflector_: Reflector::new(),
+            underlying_source_type: underlying_source_type,
+        }
+    }
+
+    pub fn new(
+        global: &GlobalScope,
+        underlying_source_type: UnderlyingSourceType,
+    ) -> DomRoot<UnderlyingSourceContainer> {
+        // TODO: setting the underlying source dict as the prototype of the
+        // `UnderlyingSourceContainer`, as it is later used as the "this" in Call_.
+        // Is this a good idea?
+        reflect_dom_object_with_proto(
+            Box::new(UnderlyingSourceContainer::new_inherited(
+                underlying_source_type,
+            )),
+            global,
+            None,
+        )
+    }
+
+    /// <https://streams.spec.whatwg.org/#dom-underlyingsource-pull>
+    #[allow(unsafe_code)]
+    pub fn call_pull_algorithm(&self, controller: Controller) -> Option<Rc<Promise>> {
+        if let UnderlyingSourceType::Js(source) = &self.underlying_source_type {
+            let global = self.global();
+            let promise = if let Some(pull) = &source.pull {
+                let cx = GlobalScope::get_cx();
+                rooted!(in(*cx) let mut this_object = ptr::null_mut::<JSObject>());
+                unsafe {
+                    source.to_jsobject(*cx, this_object.handle_mut());
+                }
+                let this_handle = this_object.handle();
+                pull.Call_(&this_handle, controller, ExceptionHandling::Report)
+                    .expect("Pull algorithm call failed")
+            } else {
+                let promise = Promise::new(&*global);
+                promise.resolve_native(&());
+                promise
+            };
+            return Some(promise);
+        }
+        // Note: other source type have no pull steps for now.
+        None
+    }
+
+    /// Does the source have all data in memory?
+    pub fn in_memory(&self) -> bool {
+        self.underlying_source_type.in_memory()
+    }
+}

components/script/dom/webidls/QueuingStrategy.webidl

@@ -0,0 +1,34 @@
+/* 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 https://mozilla.org/MPL/2.0/. */
+
+// https://streams.spec.whatwg.org/#qs
+
+dictionary QueuingStrategy {
+  unrestricted double highWaterMark;
+  QueuingStrategySize size;
+};
+
+callback QueuingStrategySize = unrestricted double (any chunk);
+
+dictionary QueuingStrategyInit {
+  required unrestricted double highWaterMark;
+};
+
+[Exposed=*]
+interface ByteLengthQueuingStrategy {
+  constructor(QueuingStrategyInit init);
+
+  readonly attribute unrestricted double highWaterMark;
+  [Throws]
+  readonly attribute Function size;
+};
+
+[Exposed=*]
+interface CountQueuingStrategy {
+  constructor(QueuingStrategyInit init);
+
+  readonly attribute unrestricted double highWaterMark;
+  [Throws]
+  readonly attribute Function size;
+};

components/script/dom/webidls/ReadableByteStreamController.webidl

@@ -0,0 +1,19 @@
+/* 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 https://mozilla.org/MPL/2.0/. */
+
+// https://streams.spec.whatwg.org/#rbs-controller-class-definition
+
+[Exposed=*]
+interface ReadableByteStreamController {
+  [Throws] // Throws on OOM
+  readonly attribute ReadableStreamBYOBRequest? byobRequest;
+  readonly attribute unrestricted double? desiredSize;
+
+  [Throws]
+  undefined close();
+  [Throws]
+  undefined enqueue(ArrayBufferView chunk);
+  [Throws]
+  undefined error(optional any e);
+};

components/script/dom/webidls/ReadableStream.webidl

@@ -2,10 +2,59 @@
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
-// This interface is entirely internal to Servo, and should not be accessible to
-// web pages.
+// https://streams.spec.whatwg.org/#readablestream
 
-[LegacyNoInterfaceObject, Exposed=(Window,Worker)]
-// Need to escape "ReadableStream" so it's treated as an identifier.
+[Exposed=*] // [Transferable] - See Bug 1562065
 interface _ReadableStream {
+  [Throws]
+  constructor(optional object underlyingSource, optional QueuingStrategy strategy = {});
+
+  // [Throws]
+  // static ReadableStream from(any asyncIterable);
+
+  readonly attribute boolean locked;
+
+  [NewObject]
+  Promise<undefined> cancel(optional any reason);
+
+  [Throws]
+  ReadableStreamReader getReader(optional ReadableStreamGetReaderOptions options = {});
+
+  // [Throws]
+  // ReadableStream pipeThrough(ReadableWritablePair transform, optional StreamPipeOptions options = {});
+
+  // [NewObject]
+  // Promise<undefined> pipeTo(WritableStream destination, optional StreamPipeOptions options = {});
+
+  // [Throws]
+  // sequence<ReadableStream> tee();
+
+  // [GenerateReturnMethod]
+  // async iterable<any>(optional ReadableStreamIteratorOptions options = {});
 };
+
+enum ReadableStreamType { "bytes" };
+
+enum ReadableStreamReaderMode { "byob" };
+
+dictionary ReadableStreamGetReaderOptions {
+  ReadableStreamReaderMode mode;
+};
+
+/*
+dictionary ReadableStreamIteratorOptions {
+  boolean preventCancel = false;
+};
+
+dictionary ReadableWritablePair {
+  required ReadableStream readable;
+  required WritableStream writable;
+};
+
+dictionary StreamPipeOptions {
+  boolean preventClose = false;
+  boolean preventAbort = false;
+  boolean preventCancel = false;
+  AbortSignal signal;
+};
+*/

components/script/dom/webidls/ReadableStreamBYOBReader.webidl

@@ -0,0 +1,19 @@
+/* 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 https://mozilla.org/MPL/2.0/. */
+
+// https://streams.spec.whatwg.org/#byob-reader-class-definition
+
+[Exposed=*]
+interface ReadableStreamBYOBReader {
+  [Throws]
+  constructor(ReadableStream stream);
+
+  [NewObject]
+  Promise<ReadableStreamReadResult> read(ArrayBufferView view);
+
+  [Throws]
+  undefined releaseLock();
+};
+ReadableStreamBYOBReader includes ReadableStreamGenericReader;
+

components/script/dom/webidls/ReadableStreamBYOBRequest.webidl

@@ -0,0 +1,15 @@
+/* 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 https://mozilla.org/MPL/2.0/. */
+
+// https://streams.spec.whatwg.org/#rs-byob-request-class-definition
+
+[Exposed=*]
+interface ReadableStreamBYOBRequest {
+  readonly attribute ArrayBufferView? view;
+
+  [Throws]
+  undefined respond([EnforceRange] unsigned long long bytesWritten);
+  [Throws]
+  undefined respondWithNewView(ArrayBufferView view);
+};

components/script/dom/webidls/ReadableStreamDefaultController.webidl

@@ -0,0 +1,23 @@
+/* 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 https://mozilla.org/MPL/2.0/. */
+
+// https://streams.spec.whatwg.org/#rs-default-controller-class-definition
+
+[Exposed=*]
+interface ReadableStreamDefaultController {
+  readonly attribute unrestricted double? desiredSize;
+
+  [Throws]
+  undefined close();
+  [Throws]
+  undefined enqueue(optional any chunk);
+  [Throws]
+  undefined error(optional any e);
+};
+
+dictionary ValueWithSize {
+ any value;
+ long size;
+};
+

components/script/dom/webidls/ReadableStreamDefaultReader.webidl

@@ -0,0 +1,34 @@
+/* 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 https://mozilla.org/MPL/2.0/. */
+
+// https://streams.spec.whatwg.org/#generic-reader-mixin-definition
+// https://streams.spec.whatwg.org/#default-reader-class-definition
+
+typedef (ReadableStreamDefaultReader or ReadableStreamBYOBReader) ReadableStreamReader;
+
+interface mixin ReadableStreamGenericReader {
+  readonly attribute Promise<undefined> closed;
+
+  [NewObject]
+  Promise<undefined> cancel(optional any reason);
+};
+
+[Exposed=*]
+interface ReadableStreamDefaultReader {
+  [Throws]
+  constructor(ReadableStream stream);
+
+  [NewObject]
+  Promise<ReadableStreamReadResult> read();
+
+  undefined releaseLock();
+};
+ReadableStreamDefaultReader includes ReadableStreamGenericReader;
+
+
+dictionary ReadableStreamReadResult {
+ any value;
+ boolean done;
+};
+

components/script/dom/webidls/UnderlyingSource.webidl

@@ -0,0 +1,21 @@
+/* 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 https://mozilla.org/MPL/2.0/. */
+
+// https://streams.spec.whatwg.org/#underlying-source-api
+
+[GenerateInit]
+dictionary UnderlyingSource {
+  UnderlyingSourceStartCallback start;
+  UnderlyingSourcePullCallback pull;
+  UnderlyingSourceCancelCallback cancel;
+  ReadableStreamType type;
+  [EnforceRange] unsigned long long autoAllocateChunkSize;
+};
+
+typedef (ReadableStreamDefaultController or ReadableByteStreamController) ReadableStreamController;
+
+callback UnderlyingSourceStartCallback = any (ReadableStreamController controller);
+callback UnderlyingSourcePullCallback = Promise<undefined> (ReadableStreamController controller);
+callback UnderlyingSourceCancelCallback = Promise<undefined> (optional any reason);
+

components/script/dom/webidls/UnderlyingSourceContainer.webidl

@@ -0,0 +1,11 @@
+/* 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 https://mozilla.org/MPL/2.0/. */
+
+// This interface is entirely internal to Servo, and should not be accessible to
+// web pages.
+
+[LegacyNoInterfaceObject, Exposed=(Window,Worker)]
+// Need to escape "ReadableStream" so it's treated as an identifier.
+interface _UnderlyingSourceContainer {
+};

python/tidy/tidy.py

@@ -85,6 +85,7 @@ WEBIDL_STANDARDS = [
     b"//encoding.spec.whatwg.org",
     b"//fetch.spec.whatwg.org",
     b"//html.spec.whatwg.org",
+    b"//streams.spec.whatwg.org",
     b"//url.spec.whatwg.org",
     b"//xhr.spec.whatwg.org",
     b"//w3c.github.io",