Hacky server streaming done

src-tauri/grpc/src/json_schema.rs

@@ -22,8 +22,7 @@ pub struct JsonSchemaEntry {
     enum_: Option<Vec<String>>,
 
     /// Don't allow any other properties in the object
-    #[serde(skip_serializing_if = "Option::is_none")]
-    additional_properties: Option<bool>,
+    additional_properties: bool,
 
     /// Set all properties to required
     #[serde(skip_serializing_if = "Option::is_none")]

src-tauri/grpc/src/lib.rs

@@ -1,8 +1,10 @@
-use prost_reflect::DynamicMessage;
+use prost::Message;
+use prost_reflect::{DynamicMessage, SerializeOptions};
 use serde::{Deserialize, Serialize};
 use serde_json::Deserializer;
-use tonic::IntoRequest;
+use tokio_stream::{Stream, StreamExt};
 use tonic::transport::Uri;
+use tonic::{IntoRequest, Response, Streaming};
 
 use crate::codec::DynamicCodec;
 use crate::proto::{fill_pool, method_desc_to_path};
@@ -11,19 +13,32 @@ mod codec;
 mod json_schema;
 mod proto;
 
+pub fn serialize_options() -> SerializeOptions {
+    SerializeOptions::new().skip_default_fields(false)
+}
+
 #[derive(Serialize, Deserialize, Debug, Default)]
+#[serde(default, rename_all = "camelCase")]
 pub struct ServiceDefinition {
     pub name: String,
     pub methods: Vec<MethodDefinition>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Default)]
+#[serde(default, rename_all = "camelCase")]
 pub struct MethodDefinition {
     pub name: String,
     pub schema: String,
+    pub client_streaming: bool,
+    pub server_streaming: bool,
 }
 
-pub async fn call(uri: &Uri, service: &str, method: &str, message_json: &str) -> String {
+pub async fn unary(
+    uri: &Uri,
+    service: &str,
+    method: &str,
+    message_json: &str,
+) -> Result<String, String> {
     let (pool, conn) = fill_pool(uri).await;
 
     let service = pool.get_service_by_name(service).unwrap();
@@ -31,7 +46,8 @@ pub async fn call(uri: &Uri, service: &str, method: &str, message_json: &str) ->
     let input_message = method.input();
 
     let mut deserializer = Deserializer::from_str(message_json);
-    let req_message = DynamicMessage::deserialize(input_message, &mut deserializer).unwrap();
+    let req_message =
+        DynamicMessage::deserialize(input_message, &mut deserializer).map_err(|e| e.to_string())?;
     deserializer.end().unwrap();
 
     let mut client = tonic::client::Grpc::new(conn);
@@ -47,10 +63,99 @@ pub async fn call(uri: &Uri, service: &str, method: &str, message_json: &str) ->
     client.ready().await.unwrap();
 
     let resp = client.unary(req, path, codec).await.unwrap();
+    let msg = resp.into_inner();
+    let response_json = serde_json::to_string_pretty(&msg).expect("json to string");
+    println!("\n---------- RECEIVING ---------------\n{}", response_json,);
+
+    Ok(response_json)
+}
+
+struct ClientStream {}
+
+impl Stream for ClientStream {
+    type Item = DynamicMessage;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        println!("poll_next");
+        todo!()
+    }
+}
+
+pub async fn client_streaming(
+    uri: &Uri,
+    service: &str,
+    method: &str,
+    message_json: &str,
+) -> Result<String, String> {
+    let (pool, conn) = fill_pool(uri).await;
+
+    let service = pool.get_service_by_name(service).unwrap();
+    let method = &service.methods().find(|m| m.name() == method).unwrap();
+    let input_message = method.input();
+
+    let mut deserializer = Deserializer::from_str(message_json);
+    let req_message =
+        DynamicMessage::deserialize(input_message, &mut deserializer).map_err(|e| e.to_string())?;
+    deserializer.end().unwrap();
+
+    let mut client = tonic::client::Grpc::new(conn);
+
+    println!(
+        "\n---------- SENDING -----------------\n{}",
+        serde_json::to_string_pretty(&req_message).expect("json")
+    );
+
+    let req = tonic::Request::new(ClientStream {});
+
+    let path = method_desc_to_path(method);
+    let codec = DynamicCodec::new(method.clone());
+    client.ready().await.unwrap();
+
+    let resp = client.client_streaming(req, path, codec).await.unwrap();
     let response_json = serde_json::to_string_pretty(&resp.into_inner()).expect("json to string");
     println!("\n---------- RECEIVING ---------------\n{}", response_json,);
 
-    response_json
+    Ok(response_json)
+}
+
+pub async fn server_streaming(
+    uri: &Uri,
+    service: &str,
+    method: &str,
+    message_json: &str,
+) -> Result<Response<Streaming<DynamicMessage>>, String> {
+    let (pool, conn) = fill_pool(uri).await;
+
+    let service = pool.get_service_by_name(service).unwrap();
+    let method = &service.methods().find(|m| m.name() == method).unwrap();
+    let input_message = method.input();
+
+    let mut deserializer = Deserializer::from_str(message_json);
+    let req_message =
+        DynamicMessage::deserialize(input_message, &mut deserializer).map_err(|e| e.to_string())?;
+    deserializer.end().unwrap();
+
+    let mut client = tonic::client::Grpc::new(conn);
+
+    println!(
+        "\n---------- SENDING -----------------\n{}",
+        serde_json::to_string_pretty(&req_message).expect("json")
+    );
+
+    let req = req_message.into_request();
+    let path = method_desc_to_path(method);
+    let codec = DynamicCodec::new(method.clone());
+    client.ready().await.unwrap();
+
+    let resp = client.server_streaming(req, path, codec).await.unwrap();
+    // let response_json = serde_json::to_string_pretty(&resp.into_inner()).expect("json to string");
+    // println!("\n---------- RECEIVING ---------------\n{}", response_json,);
+
+    // Ok(response_json)
+    Ok(resp)
 }
 
 pub async fn callable(uri: &Uri) -> Vec<ServiceDefinition> {
@@ -60,12 +165,14 @@ pub async fn callable(uri: &Uri) -> Vec<ServiceDefinition> {
         .map(|s| {
             let mut def = ServiceDefinition {
                 name: s.full_name().to_string(),
-                ..Default::default()
+                methods: vec![],
             };
             for method in s.methods() {
                 let input_message = method.input();
                 def.methods.push(MethodDefinition {
                     name: method.name().to_string(),
+                    server_streaming: method.is_server_streaming(),
+                    client_streaming: method.is_client_streaming(),
                     schema: serde_json::to_string_pretty(&json_schema::message_to_json_schema(
                         &pool,
                         input_message,

src-tauri/src/main.rs

@@ -8,32 +8,33 @@ extern crate core;
 #[macro_use]
 extern crate objc;
 
-use ::http::Uri;
 use std::collections::HashMap;
 use std::env::current_dir;
-use std::fs::{create_dir_all, read_to_string, File};
+use std::fs::{create_dir_all, File, read_to_string};
 use std::process::exit;
 use std::str::FromStr;
 
+use ::http::Uri;
 use fern::colors::ColoredLevelConfig;
-use grpc::ServiceDefinition;
+use futures::StreamExt;
 use log::{debug, error, info, warn};
 use rand::random;
 use serde::Serialize;
 use serde_json::{json, Value};
+use sqlx::{Pool, Sqlite, SqlitePool};
 use sqlx::migrate::Migrator;
 use sqlx::types::Json;
-use sqlx::{Pool, Sqlite, SqlitePool};
-#[cfg(target_os = "macos")]
-use tauri::TitleBarStyle;
 use tauri::{AppHandle, RunEvent, State, Window, WindowUrl, Wry};
 use tauri::{Manager, WindowEvent};
+#[cfg(target_os = "macos")]
+use tauri::TitleBarStyle;
 use tauri_plugin_log::{fern, LogTarget};
 use tauri_plugin_window_state::{StateFlags, WindowExt};
 use tokio::sync::Mutex;
 use tokio::time::sleep;
 use window_shadows::set_shadow;
 
+use grpc::ServiceDefinition;
 use window_ext::TrafficLightWindowExt;
 
 use crate::analytics::{AnalyticsAction, AnalyticsResource};
@@ -106,7 +107,59 @@ async fn grpc_call_unary(
     } else {
         Uri::from_str(&format!("http://{}", endpoint)).map_err(|e| e.to_string())?
     };
-    Ok(grpc::call(&uri, service, method, message).await)
+    grpc::unary(&uri, service, method, message).await
+}
+
+#[tauri::command]
+async fn grpc_client_streaming(
+    endpoint: &str,
+    service: &str,
+    method: &str,
+    message: &str,
+    // app_handle: AppHandle<Wry>,
+    // db_instance: State<'_, Mutex<Pool<Sqlite>>>,
+) -> Result<String, String> {
+    let uri = if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
+        Uri::from_str(endpoint).map_err(|e| e.to_string())?
+    } else {
+        Uri::from_str(&format!("http://{}", endpoint)).map_err(|e| e.to_string())?
+    };
+    grpc::client_streaming(&uri, service, method, message).await
+}
+
+#[tauri::command]
+async fn grpc_server_streaming(
+    endpoint: &str,
+    service: &str,
+    method: &str,
+    message: &str,
+    app_handle: AppHandle<Wry>,
+    // db_instance: State<'_, Mutex<Pool<Sqlite>>>,
+) -> Result<String, String> {
+    let uri = if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
+        Uri::from_str(endpoint).map_err(|e| e.to_string())?
+    } else {
+        Uri::from_str(&format!("http://{}", endpoint)).map_err(|e| e.to_string())?
+    };
+
+    let mut stream = grpc::server_streaming(&uri, service, method, message)
+        .await
+        .unwrap()
+        .into_inner();
+    while let Some(item) = stream.next().await {
+        match item {
+            Ok(item) => {
+                let s = serde_json::to_string(&item).unwrap();
+                emit_side_effect(&app_handle, "grpc_message", s.clone());
+                println!("GOt item: {}", s);
+            }
+            Err(e) => {
+                println!("\terror: {}", e);
+            }
+        }
+    }
+
+    Ok("foo".to_string())
 }
 
 #[tauri::command]
@@ -937,6 +990,9 @@ fn main() {
                 .level_for("reqwest", log::LevelFilter::Info)
                 .level_for("tokio_util", log::LevelFilter::Info)
                 .level_for("cookie_store", log::LevelFilter::Info)
+                .level_for("h2", log::LevelFilter::Info)
+                .level_for("tower", log::LevelFilter::Info)
+                .level_for("tonic", log::LevelFilter::Info)
                 .with_colors(ColoredLevelConfig::default())
                 .level(log::LevelFilter::Trace)
                 .build(),
@@ -1012,6 +1068,8 @@ fn main() {
             get_settings,
             get_workspace,
             grpc_call_unary,
+            grpc_client_streaming,
+            grpc_server_streaming,
             grpc_reflect,
             import_data,
             list_cookie_jars,