Add support for HTTP Streaming

plugins/http/Cargo.toml

@@ -42,6 +42,7 @@ reqwest = { version = "0.12", default-features = false }
 url = { workspace = true }
 data-url = "0.3"
 tracing = { workspace = true, optional = true }
+futures-util = "0.3.31"
 
 [features]
 default = [

plugins/http/guest-js/index.ts

@@ -26,7 +26,7 @@
  * @module
  */
 
-import { invoke } from '@tauri-apps/api/core'
+import { invoke, Channel } from '@tauri-apps/api/core'
 
 /**
  * Configuration of a proxy that a Client should pass requests to.
@@ -206,24 +206,44 @@ export async function fetch(
     rid
   })
 
-  const body = await invoke<ArrayBuffer | number[]>(
-    'plugin:http|fetch_read_body',
-    {
-      rid: responseRid
-    }
-  )
+  const channel = new Channel<number[]>()
+
+  // Create ReadableStream from channel messages
+  const stream = new ReadableStream({
+    start(controller) {
+      channel.onmessage = (arr) => {
+        const chunk = new Uint8Array(arr)
+
+        // End the stream if the chunk is empty
+        if (chunk.length === 0) {
+          controller.close()
+        } else {
+          controller.enqueue(chunk)
+        }
+      }
+
+      // Start reading body in background
+      const readPromise = invoke('plugin:http|fetch_read_body', {
+        rid: responseRid,
+        channel
+      })
 
-  const res = new Response(
-    body instanceof ArrayBuffer && body.byteLength !== 0
-      ? body
-      : body instanceof Array && body.length > 0
-        ? new Uint8Array(body)
-        : null,
-    {
-      status,
-      statusText
+      // If the promise fails, make sure the stream is closed
+      readPromise.catch((e) => {
+        console.error('error reading body', e)
+        channel.onmessage = () => {}
+        controller.error(e)
+      })
+    },
+    cancel() {
+      channel.onmessage = () => {}
     }
-  )
+  })
+
+  const res = new Response(stream, {
+    status,
+    statusText
+  })
 
   // url and headers are read only properties
   // but seems like we can set them like this

plugins/http/src/commands.rs

@@ -4,6 +4,7 @@
 
 use std::{future::Future, pin::Pin, str::FromStr, sync::Arc, time::Duration};
 
+use futures_util::StreamExt;
 use http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
 use reqwest::{redirect::Policy, NoProxy};
 use serde::{Deserialize, Serialize};
@@ -386,13 +387,34 @@ pub async fn fetch_send<R: Runtime>(
 pub(crate) async fn fetch_read_body<R: Runtime>(
     webview: Webview<R>,
     rid: ResourceId,
-) -> crate::Result<tauri::ipc::Response> {
+    channel: tauri::ipc::Channel<&[u8]>,
+) -> crate::Result<()> {
     let res = {
         let mut resources_table = webview.resources_table();
         resources_table.take::<ReqwestResponse>(rid)?
     };
+
     let res = Arc::into_inner(res).unwrap().0;
-    Ok(tauri::ipc::Response::new(res.bytes().await?.to_vec()))
+    let mut stream = res.bytes_stream();
+
+    while let Some(chunk) = stream.next().await {
+        match chunk {
+            Ok(bytes) => {
+                // Skip empty chunks
+                if !bytes.is_empty() {
+                    channel.send(&bytes)?;
+                }
+            }
+            Err(e) => {
+                return Err(e.into());
+            }
+        }
+    }
+
+    // Send an empty chunk to signal the end of the stream
+    channel.send(&[])?;
+
+    Ok(())
 }
 
 // forbidden headers per fetch spec https://fetch.spec.whatwg.org/#terminology-headers