bidi hacked!

src-tauri/Cargo.lock

@@ -6247,6 +6247,7 @@ dependencies = [
  "tauri-plugin-log",
  "tauri-plugin-window-state",
  "tokio",
+ "tokio-stream",
  "uuid",
  "window-shadows",
 ]

src-tauri/Cargo.toml

@@ -61,6 +61,7 @@ datetime = "0.5.2"
 window-shadows = "0.2.2"
 reqwest_cookie_store = "0.6.0"
 grpc = { path = "./grpc" }
+tokio-stream = "0.1.14"
 
 [features]
 # by default Tauri runs in production mode

src-tauri/grpc/src/manager.rs

@@ -3,12 +3,15 @@ use std::collections::HashMap;
 use hyper::client::HttpConnector;
 use hyper::Client;
 use hyper_rustls::HttpsConnector;
-use prost_reflect::{DescriptorPool, DynamicMessage};
+use prost_reflect::DescriptorPool;
+pub use prost_reflect::DynamicMessage;
 use serde_json::Deserializer;
 use tokio::sync::mpsc;
+use tokio_stream::wrappers::ReceiverStream;
+use tokio_stream::{Stream, StreamExt};
 use tonic::body::BoxBody;
 use tonic::transport::Uri;
-use tonic::{IntoRequest, Streaming};
+use tonic::{IntoRequest, IntoStreamingRequest, Streaming};
 
 use crate::codec::DynamicCodec;
 use crate::proto::{fill_pool, method_desc_to_path};
@@ -52,6 +55,39 @@ impl GrpcConnection {
 
         Ok(response_json)
     }
+
+    pub async fn bidi_streaming(
+        &self,
+        service: &str,
+        method: &str,
+        stream: ReceiverStream<String>,
+    ) -> Result<Streaming<DynamicMessage>> {
+        let service = self.pool.get_service_by_name(service).unwrap();
+        let method = &service.methods().find(|m| m.name() == method).unwrap();
+
+        let mut client = tonic::client::Grpc::with_origin(self.conn.clone(), self.uri.clone());
+
+        let method2 = method.clone();
+        let req = stream
+            .map(move |s| {
+                let mut deserializer = Deserializer::from_str(&s);
+                let req_message = DynamicMessage::deserialize(method2.input(), &mut deserializer)
+                    .map_err(|e| e.to_string())
+                    .unwrap();
+                deserializer.end().unwrap();
+                req_message
+            })
+            .into_streaming_request();
+        let path = method_desc_to_path(method);
+        let codec = DynamicCodec::new(method.clone());
+        client.ready().await.unwrap();
+        Ok(client
+            .streaming(req, path, codec)
+            .await
+            .map_err(|s| s.to_string())?
+            .into_inner())
+    }
+
     pub async fn server_streaming(
         &self,
         service: &str,
@@ -120,6 +156,20 @@ impl GrpcManager {
             .await
     }
 
+    pub async fn bidi_streaming(
+        &mut self,
+        id: &str,
+        uri: Uri,
+        service: &str,
+        method: &str,
+        stream: ReceiverStream<String>,
+    ) -> Result<Streaming<DynamicMessage>> {
+        println!("Bidi streaming {}", id);
+        self.connect(id, uri)
+            .await
+            .bidi_streaming(service, method, stream)
+            .await
+    }
     pub async fn connect(&mut self, id: &str, uri: Uri) -> GrpcConnection {
         let (pool, conn) = fill_pool(&uri).await;
         let connection = GrpcConnection { pool, conn, uri };

src-tauri/src/main.rs

@@ -130,6 +130,104 @@ async fn cmd_grpc_client_streaming(
     Ok(())
 }
 
+#[tauri::command]
+async fn cmd_grpc_bidi_streaming(
+    endpoint: &str,
+    service: &str,
+    method: &str,
+    app_handle: AppHandle<Wry>,
+    grpc_handle: State<'_, Mutex<GrpcManager>>,
+) -> Result<String, String> {
+    let (in_msg_tx, mut in_msg_rx) = tauri::async_runtime::channel::<String>(16);
+    let maybe_in_msg_tx = Mutex::new(Some(in_msg_tx.clone()));
+    let (cancelled_tx, mut cancelled_rx) = tokio::sync::watch::channel(false);
+
+    let uri = safe_uri(endpoint).map_err(|e| e.to_string())?;
+    let conn_id = generate_id(Some("grpc"));
+
+    let in_msg_stream = tokio_stream::wrappers::ReceiverStream::new(in_msg_rx);
+
+    let mut stream = grpc_handle
+        .lock()
+        .await
+        .bidi_streaming(&conn_id, uri, service, method, in_msg_stream)
+        .await
+        .unwrap();
+
+    #[derive(serde::Deserialize)]
+    enum GrpcMessage {
+        Message(String),
+        Commit,
+        Cancel,
+    }
+
+    let cb = {
+        let cancelled_rx = cancelled_rx.clone();
+
+        move |ev: tauri::Event| {
+            if *cancelled_rx.borrow() {
+                // Stream is cancelled
+                return;
+            }
+
+            match serde_json::from_str::<GrpcMessage>(ev.payload().unwrap()) {
+                Ok(GrpcMessage::Message(msg)) => {
+                    println!("Received message: {}", msg);
+                    in_msg_tx.try_send(msg).unwrap();
+                }
+                Ok(GrpcMessage::Commit) => {
+                    println!("Received commit");
+                    // TODO: Commit client streaming stream
+                }
+                Ok(GrpcMessage::Cancel) => {
+                    println!("Received cancel");
+                    cancelled_tx.send_replace(true);
+                }
+                Err(e) => {
+                    error!("Failed to parse gRPC message: {:?}", e);
+                }
+            }
+        }
+    };
+    let event_handler = app_handle.listen_global("grpc_message_in", cb);
+
+    let app_handle2 = app_handle.clone();
+    let grpc_listen = async move {
+        loop {
+            match stream.next().await {
+                Some(Ok(item)) => {
+                    let item = serde_json::to_string_pretty(&item).unwrap();
+                    app_handle2
+                        .emit_all("grpc_message", item)
+                        .expect("Failed to emit");
+                }
+                Some(Err(e)) => {
+                    error!("gRPC stream error: {:?}", e);
+                    // TODO: Handle error
+                }
+                None => {
+                    info!("gRPC stream closed by sender");
+                    break;
+                }
+            }
+        }
+    };
+
+    tauri::async_runtime::spawn(async move {
+        tokio::select! {
+            _ = grpc_listen => {
+                debug!("gRPC listen finished");
+            },
+            _ = cancelled_rx.changed() => {
+                debug!("gRPC connection cancelled");
+            },
+        }
+        app_handle.unlisten(event_handler);
+    });
+
+    Ok(conn_id)
+}
+
 #[tauri::command]
 async fn cmd_grpc_server_streaming(
     endpoint: &str,
@@ -1119,6 +1217,7 @@ fn main() {
             cmd_grpc_call_unary,
             cmd_grpc_client_streaming,
             cmd_grpc_server_streaming,
+            cmd_grpc_bidi_streaming,
             cmd_grpc_reflect,
             cmd_import_data,
             cmd_list_cookie_jars,