HTTP add stream support

plugins/http/guest-js/index.ts

@@ -26,7 +26,7 @@
  * @module
  */
 
-import { invoke } from '@tauri-apps/api/core'
+import { Channel, invoke } from '@tauri-apps/api/core'
 
 /**
  * Configuration of a proxy that a Client should pass requests to.
@@ -106,6 +106,20 @@ export interface DangerousSettings {
   acceptInvalidHostnames?: boolean
 }
 
+/**
+ * Stream Packet for IPC
+ */
+export interface StreamMessage {
+  /**
+   * The chunk - an array of bytes sent from Rust.
+   */
+  value?: ArrayBuffer | number[]
+  /**
+   * Is the stream done.
+   */
+  done: boolean
+}
+
 const ERROR_REQUEST_CANCELLED = 'Request canceled'
 
 /**
@@ -186,6 +200,19 @@ export async function fetch(
     throw new Error(ERROR_REQUEST_CANCELLED)
   }
 
+  const streamChannel = new Channel<StreamMessage>()
+
+  const readableStreamBody = new ReadableStream({
+    start: (controller) => {
+      streamChannel.onmessage = (res: StreamMessage) => {
+        // close early if aborted
+        if (signal?.aborted) controller.error(ERROR_REQUEST_CANCELLED)
+        if (res.done) controller.close()
+        controller.enqueue(res.value)
+      }
+    }
+  })
+
   const rid = await invoke<number>('plugin:http|fetch', {
     clientConfig: {
       method: req.method,
@@ -196,7 +223,8 @@ export async function fetch(
       connectTimeout,
       proxy,
       danger
-    }
+    },
+    streamChannel
   })
 
   const abort = () => invoke('plugin:http|fetch_cancel', { rid })
@@ -223,30 +251,15 @@ export async function fetch(
     status,
     statusText,
     url,
-    headers: responseHeaders,
-    rid: responseRid
+    headers: responseHeaders
   } = await invoke<FetchSendResponse>('plugin:http|fetch_send', {
     rid
   })
 
-  const body = await invoke<ArrayBuffer | number[]>(
-    'plugin:http|fetch_read_body',
-    {
-      rid: responseRid
-    }
-  )
-
-  const res = new Response(
-    body instanceof ArrayBuffer && body.byteLength !== 0
-      ? body
-      : body instanceof Array && body.length > 0
-        ? new Uint8Array(body)
-        : null,
-    {
-      status,
-      statusText
-    }
-  )
+  const res = new Response(readableStreamBody, {
+    status,
+    statusText
+  })
 
   // url and headers are read only properties
   // but seems like we can set them like this

plugins/http/src/commands.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-License-Identifier: MIT
 
+
 use std::{future::Future, pin::Pin, str::FromStr, sync::Arc, time::Duration};
 
 use http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
@@ -10,7 +11,7 @@ use serde::{Deserialize, Serialize};
 use tauri::{
     async_runtime::Mutex,
     command,
-    ipc::{CommandScope, GlobalScope},
+    ipc::{Channel, CommandScope, GlobalScope},
     Manager, ResourceId, ResourceTable, Runtime, State, Webview,
 };
 use tokio::sync::oneshot::{channel, Receiver, Sender};
@@ -22,6 +23,8 @@ use crate::{
 
 const HTTP_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
 
+// reqwest::Response is never read, but might be needed for future use.
+#[allow(dead_code)]
 struct ReqwestResponse(reqwest::Response);
 impl tauri::Resource for ReqwestResponse {}
 
@@ -126,6 +129,12 @@ pub struct BasicAuth {
     password: String,
 }
 
+#[derive(Clone, Serialize)]
+pub struct StreamMessage {
+    value: Option<Vec<u8>>,
+    done: bool,
+}
+
 #[inline]
 fn proxy_creator(
     url_or_config: UrlOrConfig,
@@ -181,6 +190,7 @@ pub async fn fetch<R: Runtime>(
     client_config: ClientConfig,
     command_scope: CommandScope<Entry>,
     global_scope: GlobalScope<Entry>,
+    stream_channel: Channel<StreamMessage>
 ) -> crate::Result<ResourceId> {
     let ClientConfig {
         method,
@@ -314,7 +324,24 @@ pub async fn fetch<R: Runtime>(
                 #[cfg(feature = "tracing")]
                 tracing::trace!("{:?}", request);
 
-                let fut = async move { request.send().await.map_err(Into::into) };
+                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(StreamMessage{
+                            value: Some(chunk.to_vec()),
+                            done: false,
+                        })?;
+                    }
+
+                    stream_channel.send(StreamMessage { value: None, done: true })?;
+
+                    // return that response
+                    Ok(res)
+                };
+
+
                 let mut resources_table = webview.resources_table();
                 let rid = resources_table.add_request(Box::pin(fut));
 
@@ -410,19 +437,6 @@ pub async fn fetch_send<R: Runtime>(
     })
 }
 
-#[tauri::command]
-pub(crate) async fn fetch_read_body<R: Runtime>(
-    webview: Webview<R>,
-    rid: ResourceId,
-) -> crate::Result<tauri::ipc::Response> {
-    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()))
-}
-
 // 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,8 +36,7 @@ pub fn init<R: Runtime>() -> TauriPlugin<R> {
         .invoke_handler(tauri::generate_handler![
             commands::fetch,
             commands::fetch_cancel,
-            commands::fetch_send,
-            commands::fetch_read_body,
+            commands::fetch_send
         ])
         .build()
 }