feat(fetch): implement fetch body streams

Closes #946

Implements Response.body and Request.body as ReadableStream per WHATWG Fetch spec:

- Response.body returns ReadableStream<Uint8Array> for streaming response data
- Request.body returns ReadableStream<Uint8Array> for request bodies
- Byte stream controller (type: "bytes") with BYOB reader support
- Response.clone() uses tee() to allow cloning after body access
- fetch() accepts ReadableStream as request body for streaming uploads
- Compressed responses (gzip/deflate/brotli) stream decompressed data
- pipeTo() support for piping body streams to WritableStream

Added 42 new tests covering fetch body stream functionality.

Author: chessbyte

API.md

@@ -535,8 +535,7 @@ export class XMLParser(options?: XmlParserOptions){
 > There are some differences with the [WHATWG standard](https://fetch.spec.whatwg.org). Mainly browser specific behavior is removed:
 >
 > - `keepalive` is always true
-> - `request.body` can only be `string`, `Array`, `ArrayBuffer` or `Uint8Array`
-> - `response.body` returns `null`. Use `response.text()`, `response.json()` etc
+> - `request.body` can only be `string`, `Array`, `ArrayBuffer`, `Uint8Array`, `Blob`, or `ReadableStream`
 > - `mode`, `credentials`, `referrerPolicy`, `priority`, `cache` is not available/applicable
 
 ## FILEAPI

Cargo.lock

@@ -36,6 +36,7 @@ llrt_context = { version = "0.7.0-beta", path = "../../libs/llrt_context" }
 llrt_encoding = { version = "0.7.0-beta", path = "../../libs/llrt_encoding" }
 llrt_http = { version = "0.7.0-beta", default-features = false, path = "../llrt_http" }
 llrt_json = { version = "0.7.0-beta", path = "../../libs/llrt_json" }
+llrt_stream_web = { version = "0.7.0-beta", path = "../llrt_stream_web" }
 llrt_url = { version = "0.7.0-beta", path = "../llrt_url" }
 llrt_utils = { version = "0.7.0-beta", path = "../../libs/llrt_utils", default-features = false }
 once_cell = { version = "1", features = ["std"], default-features = false }

modules/llrt_fetch/Cargo.toml

@@ -25,6 +25,7 @@ use rquickjs::{
 use tokio::{select, sync::Semaphore};
 
 use super::{
+    body_stream::read_all_bytes_from_stream,
     headers::{Headers, HeadersGuard},
     response::Response,
     security::ensure_url_access,
@@ -85,6 +86,18 @@ pub fn init(global_client: HyperClient, globals: &Object) -> Result<()> {
 
                 ensure_url_access(&ctx, &uri)?;
 
+                // Convert stream body to bytes if needed (must be done before request loop)
+                let body_bytes: Option<BodyBytes> = match options.body {
+                    Some(FetchBody::Bytes(bytes)) => Some(bytes),
+                    Some(FetchBody::Stream(stream)) => {
+                        let bytes = read_all_bytes_from_stream(&ctx, stream).await?;
+                        let typed_array = bytes_to_typed_array(ctx.clone(), &bytes)?;
+                        let object_bytes = ObjectBytes::from(&ctx, &typed_array)?;
+                        Some(BodyBytes::new(ctx.clone(), object_bytes)?)
+                    },
+                    None => None,
+                };
+
                 let mut redirect_count = 0;
                 let mut response_status = 0;
                 let (res, guard) = loop {
@@ -93,7 +106,7 @@ pub fn init(global_client: HyperClient, globals: &Object) -> Result<()> {
                         &method,
                         &uri,
                         options.headers.as_ref(),
-                        options.body.as_ref(),
+                        body_bytes.as_ref(),
                         &response_status,
                         &initial_uri,
                     )?;
@@ -331,11 +344,16 @@ impl<'js> BodyBytes<'js> {
     }
 }
 
+enum FetchBody<'js> {
+    Bytes(BodyBytes<'js>),
+    Stream(llrt_stream_web::ReadableStreamClass<'js>),
+}
+
 struct FetchOptions<'js> {
     method: hyper::Method,
     url: String,
     headers: Option<Headers>,
-    body: Option<BodyBytes<'js>>,
+    body: Option<FetchBody<'js>>,
     abort_receiver: Option<mc_oneshot::Receiver<Value<'js>>>,
     redirect: String,
     agent: Option<Class<'js, Agent>>,
@@ -395,14 +413,22 @@ fn get_fetch_options<'js>(
         if let Some(body_opt) =
             get_option::<Value>("body", arg_opts.as_ref(), resource_opts.as_ref())?
         {
-            let bytes = if let Ok(blob) = Class::<Blob>::from_value(&body_opt) {
-                let blob = blob.borrow();
-                let typed_array = bytes_to_typed_array(ctx.clone(), &blob.get_bytes())?;
-                ObjectBytes::from(ctx, &typed_array)?
+            // Check if body is a ReadableStream first
+            if let Some(stream) = body_opt
+                .as_object()
+                .and_then(Class::<llrt_stream_web::ReadableStreamStruct>::from_object)
+            {
+                body = Some(FetchBody::Stream(stream));
             } else {
-                ObjectBytes::from(ctx, &body_opt)?
-            };
-            body = Some(BodyBytes::new(ctx.clone(), bytes)?);
+                let bytes = if let Ok(blob) = Class::<Blob>::from_value(&body_opt) {
+                    let blob = blob.borrow();
+                    let typed_array = bytes_to_typed_array(ctx.clone(), &blob.get_bytes())?;
+                    ObjectBytes::from(ctx, &typed_array)?
+                } else {
+                    ObjectBytes::from(ctx, &body_opt)?
+                };
+                body = Some(FetchBody::Bytes(BodyBytes::new(ctx.clone(), bytes)?));
+            }
         }
 
         if let Some(url_opt) =

modules/llrt_fetch/src/body_stream.rs

@@ -14,6 +14,7 @@ pub use self::security::{get_allow_list, get_deny_list, set_allow_list, set_deny
 use self::{form_data::FormData, headers::Headers, request::Request, response::Response};
 
 mod body;
+mod body_stream;
 pub mod fetch;
 pub mod form_data;
 pub mod headers;

modules/llrt_fetch/src/fetch.rs

@@ -12,6 +12,9 @@ use rquickjs::{
     IntoJs, Null, Object, Result, TypedArray, Value,
 };
 
+use crate::body_stream::{create_value_stream, read_all_bytes_from_stream};
+use llrt_stream_web::ReadableStreamClass;
+
 use super::{
     headers::{Headers, HeadersGuard, HEADERS_KEY_CONTENT_TYPE},
     strip_bom, Blob, FormData, MIME_TYPE_FORM_DATA, MIME_TYPE_FORM_URLENCODED,
@@ -56,6 +59,7 @@ impl RequestMode {
 #[derive(rquickjs::JsLifetime)]
 enum BodyVariant<'js> {
     Provided(Option<Value<'js>>),
+    Stream(Option<ReadableStreamClass<'js>>),
     Empty,
 }
 
@@ -79,8 +83,10 @@ impl<'js> Trace<'js> for Request<'js> {
         }
         let body = self.body.read().unwrap();
         let body = &*body;
-        if let BodyVariant::Provided(Some(body)) = body {
-            body.trace(tracer);
+        match body {
+            BodyVariant::Provided(Some(body)) => body.trace(tracer),
+            BodyVariant::Stream(Some(stream)) => stream.trace(tracer),
+            _ => {},
         }
     }
 }
@@ -142,13 +148,32 @@ impl<'js> Request<'js> {
         stringify!(Request)
     }
 
-    //TODO should implement readable stream
+    /// Returns the body as a ReadableStream.
+    ///
+    /// Per WHATWG Fetch specification:
+    /// - Returns null if the body is null (empty)
+    /// - Returns a ReadableStream that yields Uint8Array chunks
     #[qjs(get)]
     fn body(&self, ctx: Ctx<'js>) -> Result<Value<'js>> {
-        let body = self.body.read().unwrap();
-        let body = &*body;
+        let mut body_guard = self.body.write().unwrap();
+        let body = &mut *body_guard;
         match body {
-            BodyVariant::Provided(value) => value.into_js(&ctx),
+            BodyVariant::Provided(provided) => {
+                if let Some(value) = provided.take() {
+                    create_value_stream(&ctx, value)
+                } else {
+                    // Body already consumed - return null
+                    Ok(Value::new_null(ctx))
+                }
+            },
+            BodyVariant::Stream(stream) => {
+                if let Some(s) = stream.take() {
+                    Ok(s.into_value())
+                } else {
+                    // Stream already consumed - return null
+                    Ok(Value::new_null(ctx))
+                }
+            },
             BodyVariant::Empty => Null.into_js(&ctx),
         }
     }
@@ -169,6 +194,7 @@ impl<'js> Request<'js> {
         let body = &*body;
         match body {
             BodyVariant::Provided(value) => value.is_none(),
+            BodyVariant::Stream(stream) => stream.is_none(),
             BodyVariant::Empty => false,
         }
     }
@@ -252,6 +278,7 @@ impl<'js> Request<'js> {
         let body = &*body;
         let body = match body {
             BodyVariant::Provided(provided) => BodyVariant::Provided(provided.clone()),
+            BodyVariant::Stream(stream) => BodyVariant::Stream(stream.clone()),
             BodyVariant::Empty => BodyVariant::Empty,
         };
 
@@ -288,6 +315,13 @@ impl<'js> Request<'js> {
                     bytes.as_bytes(ctx)?.to_vec()
                 }
             },
+            BodyVariant::Stream(stream) => {
+                let stream = stream
+                    .take()
+                    .ok_or(Exception::throw_message(ctx, "Already read"))?;
+                drop(body_guard);
+                read_all_bytes_from_stream(ctx, stream).await?
+            },
             BodyVariant::Empty => return Ok(None),
         };
 
@@ -361,7 +395,12 @@ fn assign_request<'js>(request: &mut Request<'js>, ctx: Ctx<'js>, obj: &Object<'
                 content_type = Some(MIME_TYPE_TEXT.into());
                 BodyVariant::Provided(Some(body))
             } else if let Some(obj) = body.as_object() {
-                if let Some(blob) = Class::<Blob>::from_object(obj) {
+                if let Some(stream) =
+                    Class::<llrt_stream_web::ReadableStreamStruct>::from_object(obj)
+                {
+                    // ReadableStream body - store as Stream variant
+                    BodyVariant::Stream(Some(stream))
+                } else if let Some(blob) = Class::<Blob>::from_object(obj) {
                     let blob = blob.borrow();
                     if !blob.mime_type().is_empty() {
                         content_type = Some(blob.mime_type());