fix(http): Fixes Stream Bug

.changes/http-stream-support.md

@@ -0,0 +1,6 @@
+---
+"http": minor
+"http-js": minor
+---
+
+Fix HTTP stream not streaming the body.

plugins/http/guest-js/index.ts

@@ -106,7 +106,7 @@ export interface DangerousSettings {
   acceptInvalidHostnames?: boolean
 }
 
-const ERROR_REQUEST_CANCELLED = 'Request canceled'
+const ERROR_REQUEST_CANCELLED = 'Request cancelled'
 
 /**
  * Fetch a resource from the network. It returns a `Promise` that resolves to the
@@ -186,35 +186,6 @@ export async function fetch(
     throw new Error(ERROR_REQUEST_CANCELLED)
   }
 
-  const streamChannel = new Channel<ArrayBuffer | number[]>()
-
-  const readableStreamBody = new ReadableStream({
-    start: (controller) => {
-      streamChannel.onmessage = (res: ArrayBuffer | number[]) => {
-        // close early if aborted
-        if (signal?.aborted) {
-          controller.error(ERROR_REQUEST_CANCELLED)
-          controller.close()
-          return
-        }
-
-        // close when the signal to close (an empty chunk)
-        // is sent from the IPC.
-        if (
-          res instanceof ArrayBuffer ? res.byteLength == 0 : res.length == 0
-        ) {
-          controller.close()
-          return
-        }
-
-        // the content conversion (like .text(), .json(), etc.) in Response
-        // must have Uint8Array as its content, else it will
-        // have untraceable error that's hard to debug.
-        controller.enqueue(new Uint8Array(res))
-      }
-    }
-  })
-
   const rid = await invoke<number>('plugin:http|fetch', {
     clientConfig: {
       method: req.method,
@@ -225,8 +196,7 @@ export async function fetch(
       connectTimeout,
       proxy,
       danger
-    },
-    streamChannel
+    }
   })
 
   const abort = () => invoke('plugin:http|fetch_cancel', { rid })
@@ -253,11 +223,47 @@ export async function fetch(
     status,
     statusText,
     url,
-    headers: responseHeaders
+    headers: responseHeaders,
+    rid: responseRid
   } = await invoke<FetchSendResponse>('plugin:http|fetch_send', {
     rid
   })
 
+  const readableStreamBody = new ReadableStream({
+    start: (controller) => {
+      const streamChannel = new Channel<ArrayBuffer | number[]>()
+      streamChannel.onmessage = (res: ArrayBuffer | number[]) => {
+        // close early if aborted
+        if (signal?.aborted) {
+          controller.error(ERROR_REQUEST_CANCELLED)
+          return
+        }
+
+        // close when the signal to close (an empty chunk)
+        // is sent from the IPC.
+        if (
+          res instanceof ArrayBuffer ? res.byteLength == 0 : res.length == 0
+        ) {
+          controller.close()
+          return
+        }
+
+        // the content conversion (like .text(), .json(), etc.) in Response
+        // must have Uint8Array as its content, else it will
+        // have untraceable error that's hard to debug.
+        controller.enqueue(new Uint8Array(res))
+      }
+
+      // run a non-blocking body stream fetch
+      invoke('plugin:http|fetch_read_body', {
+        rid: responseRid,
+        streamChannel
+      }).catch((e) => {
+        controller.error(e)
+      })
+    }
+  })
+
   const res = new Response(readableStreamBody, {
     status,
     statusText

plugins/http/src/commands.rs

@@ -22,6 +22,9 @@ use crate::{
 
 const HTTP_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
 
+struct ReqwestResponse(reqwest::Response);
+impl tauri::Resource for ReqwestResponse {}
+
 type CancelableResponseResult = Result<reqwest::Response>;
 type CancelableResponseFuture =
     Pin<Box<dyn Future<Output = CancelableResponseResult> + Send + Sync>>;
@@ -178,7 +181,6 @@ pub async fn fetch<R: Runtime>(
     client_config: ClientConfig,
     command_scope: CommandScope<Entry>,
     global_scope: GlobalScope<Entry>,
-    stream_channel: Channel<tauri::ipc::InvokeResponseBody>,
 ) -> crate::Result<ResourceId> {
     let ClientConfig {
         method,
@@ -312,20 +314,7 @@ pub async fn fetch<R: Runtime>(
                 #[cfg(feature = "tracing")]
                 tracing::trace!("{:?}", request);
 
-                let fut = async move {
-                    let mut res = request.send().await?;
-
-                    // send response through IPC channel
-                    while let Some(chunk) = res.chunk().await? {
-                        stream_channel.send(tauri::ipc::InvokeResponseBody::Raw(chunk.to_vec()))?;
-                    }
-
-                    // send empty vector when done
-                    stream_channel.send(tauri::ipc::InvokeResponseBody::Raw(Vec::new()))?;
-
-                    // return that response
-                    Ok(res)
-                };
+                let fut = async move { request.send().await.map_err(Into::into) };
 
                 let mut resources_table = webview.resources_table();
                 let rid = resources_table.add_request(Box::pin(fut));
@@ -370,7 +359,7 @@ pub fn fetch_cancel<R: Runtime>(webview: Webview<R>, rid: ResourceId) -> crate::
     Ok(())
 }
 
-#[tauri::command]
+#[command]
 pub async fn fetch_send<R: Runtime>(
     webview: Webview<R>,
     rid: ResourceId,
@@ -410,6 +399,9 @@ pub async fn fetch_send<R: Runtime>(
         ));
     }
 
+    let mut resources_table = webview.resources_table();
+    let rid = resources_table.add(ReqwestResponse(res));
+
     Ok(FetchResponse {
         status: status.as_u16(),
         status_text: status.canonical_reason().unwrap_or_default().to_string(),
@@ -419,6 +411,30 @@ pub async fn fetch_send<R: Runtime>(
     })
 }
 
+#[command]
+pub async fn fetch_read_body<R: Runtime>(
+    webview: Webview<R>,
+    rid: ResourceId,
+    stream_channel: Channel<tauri::ipc::InvokeResponseBody>,
+) -> crate::Result<()> {
+    let res = {
+        let mut resources_table = webview.resources_table();
+        resources_table.take::<ReqwestResponse>(rid)?
+    };
+
+    let mut res = Arc::into_inner(res).unwrap().0;
+
+    // send response through IPC channel
+    while let Some(chunk) = res.chunk().await? {
+        stream_channel.send(tauri::ipc::InvokeResponseBody::Raw(chunk.to_vec()))?;
+    }
+
+    // send empty vector when done
+    stream_channel.send(tauri::ipc::InvokeResponseBody::Raw(Vec::new()))?;
+
+    Ok(())
+}
+
 // forbidden headers per fetch spec https://fetch.spec.whatwg.org/#terminology-headers
 #[cfg(not(feature = "unsafe-headers"))]
 fn is_unsafe_header(header: &HeaderName) -> bool {

plugins/http/src/lib.rs

@@ -36,7 +36,8 @@ pub fn init<R: Runtime>() -> TauriPlugin<R> {
         .invoke_handler(tauri::generate_handler![
             commands::fetch,
             commands::fetch_cancel,
-            commands::fetch_send
+            commands::fetch_send,
+            commands::fetch_read_body
         ])
         .build()
 }