diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 68019756..4c6f07e1 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -1859,20 +1859,6 @@ dependencies = [ "windows-sys 0.48.0", ] -[[package]] -name = "html5ever" -version = "0.25.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e5c13fb08e5d4dfc151ee5e88bae63f7773d61852f3bdc73c9f4b9e1bde03148" -dependencies = [ - "log", - "mac", - "markup5ever 0.10.1", - "proc-macro2", - "quote", - "syn 1.0.109", -] - [[package]] name = "html5ever" version = "0.26.0" @@ -1881,7 +1867,7 @@ checksum = "bea68cab48b8459f17cf1c944c67ddc572d272d9f2b274140f223ecb1da4a3b7" dependencies = [ "log", "mac", - "markup5ever 0.11.0", + "markup5ever", "proc-macro2", "quote", "syn 1.0.109", @@ -2359,18 +2345,6 @@ dependencies = [ "treediff", ] -[[package]] -name = "kuchiki" -version = "0.8.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ea8e9c6e031377cff82ee3001dc8026cdf431ed4e2e6b51f98ab8c73484a358" -dependencies = [ - "cssparser", - "html5ever 0.25.2", - "matches", - "selectors", -] - [[package]] name = "kuchikiki" version = "0.8.2" @@ -2378,7 +2352,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f29e4755b7b995046f510a7520c42b2fed58b77bd94d5a87a8eb43d2fd126da8" dependencies = [ "cssparser", - "html5ever 0.26.0", + "html5ever", "indexmap 1.9.3", "matches", "selectors", @@ -2506,20 +2480,6 @@ dependencies = [ "libc", ] -[[package]] -name = "markup5ever" -version = "0.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a24f40fb03852d1cdd84330cddcaf98e9ec08a7b7768e952fad3b4cf048ec8fd" -dependencies = [ - "log", - "phf 0.8.0", - "phf_codegen 0.8.0", - "string_cache", - "string_cache_codegen", - "tendril", -] - [[package]] name = "markup5ever" version = "0.11.0" @@ -4717,9 +4677,9 @@ checksum = "14c39fd04924ca3a864207c66fc2cd7d22d7c016007f9ce846cbb9326331930a" [[package]] name = "tauri" -version = "1.5.2" +version = "1.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9bfe673cf125ef364d6f56b15e8ce7537d9ca7e4dae1cf6fbbdeed2e024db3d9" +checksum = "fd27c04b9543776a972c86ccf70660b517ecabbeced9fb58d8b961a13ad129af" dependencies = [ "anyhow", "base64 0.21.5", @@ -4792,9 +4752,9 @@ dependencies = [ [[package]] name = "tauri-codegen" -version = "1.4.1" +version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b3475e55acec0b4a50fb96435f19631fb58cbcd31923e1a213de5c382536bbb" +checksum = "a1554c5857f65dbc377cefb6b97c8ac77b1cb2a90d30d3448114d5d6b48a77fc" dependencies = [ "base64 0.21.5", "brotli", @@ -4818,9 +4778,9 @@ dependencies = [ [[package]] name = "tauri-macros" -version = "1.4.1" +version = "1.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "613740228de92d9196b795ac455091d3a5fbdac2654abb8bb07d010b62ab43af" +checksum = "277abf361a3a6993ec16bcbb179de0d6518009b851090a01adfea12ac89fa875" dependencies = [ "heck 0.4.1", "proc-macro2", @@ -4861,9 +4821,9 @@ dependencies = [ [[package]] name = "tauri-runtime" -version = "0.14.1" +version = "0.14.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "07f8e9e53e00e9f41212c115749e87d5cd2a9eebccafca77a19722eeecd56d43" +checksum = "cf2d0652aa2891ff3e9caa2401405257ea29ab8372cce01f186a5825f1bd0e76" dependencies = [ "gtk", "http", @@ -4882,9 +4842,9 @@ dependencies = [ [[package]] name = "tauri-runtime-wry" -version = "0.14.1" +version = "0.14.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8141d72b6b65f2008911e9ef5b98a68d1e3413b7a1464e8f85eb3673bb19a895" +checksum = "6cae61fbc731f690a4899681c9052dde6d05b159b44563ace8186fc1bfb7d158" dependencies = [ "cocoa 0.24.1", "gtk", @@ -4911,7 +4871,7 @@ dependencies = [ "dunce", "glob", "heck 0.4.1", - "html5ever 0.26.0", + "html5ever", "infer", "json-patch", "kuchikiki", @@ -6193,9 +6153,9 @@ checksum = "dad7bb64b8ef9c0aa27b6da38b452b0ee9fd82beaf276a87dd796fb55cbae14e" [[package]] name = "wry" -version = "0.24.4" +version = "0.24.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88ef04bdad49eba2e01f06e53688c8413bd6a87b0bc14b72284465cf96e3578e" +checksum = "6ad85d0e067359e409fcb88903c3eac817c392e5d638258abfb3da5ad8ba6fc4" dependencies = [ "base64 0.13.1", "block", @@ -6207,9 +6167,9 @@ dependencies = [ "gio", "glib", "gtk", - "html5ever 0.25.2", + "html5ever", "http", - "kuchiki", + "kuchikiki", "libc", "log", "objc", diff --git a/src-tauri/grpc/src/manager.rs b/src-tauri/grpc/src/manager.rs index a3ec03e0..3debec82 100644 --- a/src-tauri/grpc/src/manager.rs +++ b/src-tauri/grpc/src/manager.rs @@ -78,7 +78,6 @@ impl GrpcConnection { let path = method_desc_to_path(method); let codec = DynamicCodec::new(method.clone()); client.ready().await.unwrap(); - Ok(client .server_streaming(req, path, codec) .await diff --git a/src-tauri/grpc/src/proto.rs b/src-tauri/grpc/src/proto.rs index e59dfbd9..9a12df25 100644 --- a/src-tauri/grpc/src/proto.rs +++ b/src-tauri/grpc/src/proto.rs @@ -36,12 +36,6 @@ pub async fn fill_pool( .http2_only(true) .build(connector); - println!( - "URI uri={} host={:?} authority={:?}", - uri, - uri.host(), - uri.authority() - ); let mut client = ServerReflectionClient::with_origin(transport.clone(), uri.clone()); let services = list_services(&mut client).await; diff --git a/src-tauri/src/http.rs b/src-tauri/src/http.rs index d9b677e7..307166a2 100644 --- a/src-tauri/src/http.rs +++ b/src-tauri/src/http.rs @@ -1,5 +1,5 @@ -use std::fs::{create_dir_all, File}; use std::fs; +use std::fs::{create_dir_all, File}; use std::io::Write; use std::path::PathBuf; use std::str::FromStr; @@ -7,19 +7,19 @@ use std::sync::Arc; use std::time::Duration; use base64::Engine; -use http::{HeaderMap, HeaderName, HeaderValue, Method}; use http::header::{ACCEPT, USER_AGENT}; +use http::{HeaderMap, HeaderName, HeaderValue, Method}; use log::{error, info, warn}; -use reqwest::{multipart, Url}; use reqwest::redirect::Policy; -use sqlx::{Pool, Sqlite}; +use reqwest::{multipart, Url}; use sqlx::types::{Json, JsonValue}; +use sqlx::{Pool, Sqlite}; use tauri::{AppHandle, Wry}; use crate::{emit_side_effect, models, render, response_err}; pub async fn send_http_request( - app_handle: &AppHandle, + app_handle: AppHandle, db: &Pool, request: models::HttpRequest, response: &models::HttpResponse, @@ -88,7 +88,7 @@ pub async fn send_http_request( let url = match Url::from_str(url_string.as_str()) { Ok(u) => u, Err(e) => { - return response_err(response, e.to_string(), app_handle, db).await; + return response_err(response, e.to_string(), app_handle.clone(), db).await; } }; @@ -293,7 +293,7 @@ pub async fn send_http_request( let sendable_req = match request_builder.build() { Ok(r) => r, Err(e) => { - return response_err(response, e.to_string(), app_handle, db).await; + return response_err(response, e.to_string(), app_handle.clone(), db).await; } }; @@ -366,7 +366,7 @@ pub async fn send_http_request( .await .expect("Failed to update response"); if !request.id.is_empty() { - emit_side_effect(app_handle, "updated_model", &response); + emit_side_effect(app_handle.clone(), "updated_model", &response); } // Copy response to download path, if specified diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index ba06b93a..651b02fd 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -95,14 +95,6 @@ async fn cmd_grpc_reflect(endpoint: &str) -> Result, Stri Ok(grpc::callable(&uri).await) } -async fn cmd_grpc_cancel( - id: &str, - grpc_handle: State<'_, Mutex>, -) -> Result<(), String> { - // grpc_handle.lock().await.cancel(id).await.unwrap() - Ok(()) -} - #[tauri::command] async fn cmd_grpc_call_unary( endpoint: &str, @@ -147,6 +139,8 @@ async fn cmd_grpc_server_streaming( app_handle: AppHandle, grpc_handle: State<'_, Mutex>, ) -> Result { + 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")); @@ -157,44 +151,76 @@ async fn cmd_grpc_server_streaming( .await .unwrap(); - loop { - match stream.message().await { - Ok(Some(item)) => { - let item = serde_json::to_string_pretty(&item).unwrap(); - println!("GOT MESSAGE {:?}", item); - println!("Sending message {}", item); - emit_side_effect(&app_handle, "grpc_message", item); - } - Ok(None) => { - // sleep for a bit - println!("NO MESSAGE YET"); - sleep(std::time::Duration::from_millis(100)).await; - } - Err(e) => { - return Err(e.to_string()); - } - } + #[derive(serde::Deserialize)] + enum GrpcMessage { + Message(String), + Commit, + Cancel, } - // while let Some(item) = stream.message() { - // println!("GOT MESSAGE"); - // // if grpc_handle.lock().await.is_cancelled(&conn_id) { - // // break; - // // } - // - // match item { - // Ok(item) => { - // let item = serde_json::to_string_pretty(&item).unwrap(); - // println!("Sending message {}", item); - // emit_side_effect(&app_handle, "grpc_message", item); - // } - // Err(e) => println!("\terror: {}", e), - // } - // // let foo = stream.trailers().await.unwrap(); - // break; - // } + 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::(ev.payload().unwrap()) { + Ok(GrpcMessage::Message(msg)) => { + println!("Received message: {}", msg); + } + 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); + }); - println!("DONE"); Ok(conn_id) } @@ -228,7 +254,7 @@ async fn cmd_send_ephemeral_request( // let cookie_jar_id2 = cookie_jar_id.unwrap_or("").to_string(); send_http_request( - &app_handle, + app_handle, db, request, &response, @@ -381,7 +407,7 @@ async fn cmd_export_data( #[tauri::command] async fn cmd_send_request( - window: Window, + app_handle: AppHandle, db_state: State<'_, Mutex>>, request_id: &str, environment_id: Option<&str>, @@ -389,7 +415,6 @@ async fn cmd_send_request( download_dir: Option<&str>, ) -> Result { let db = &*db_state.lock().await; - let app_handle = window.app_handle(); let request = get_request(db, request_id) .await @@ -436,10 +461,10 @@ async fn cmd_send_request( None }; - emit_side_effect(&app_handle, "created_model", response.clone()); + emit_side_effect(app_handle.clone(), "created_model", response.clone()); send_http_request( - &app_handle, + app_handle, db, request.clone(), &response, @@ -453,7 +478,7 @@ async fn cmd_send_request( async fn response_err( response: &HttpResponse, error: String, - app_handle: &AppHandle, + app_handle: AppHandle, db: &Pool, ) -> Result { let mut response = response.clone(); @@ -1000,24 +1025,25 @@ async fn cmd_check_for_updates( fn main() { tauri::Builder::default() + .plugin(tauri_plugin_window_state::Builder::default().build()) .plugin( tauri_plugin_log::Builder::default() .targets([LogTarget::LogDir, LogTarget::Stdout, LogTarget::Webview]) - .level_for("tao", log::LevelFilter::Info) - .level_for("sqlx", log::LevelFilter::Warn) - .level_for("hyper", log::LevelFilter::Info) - .level_for("tracing", log::LevelFilter::Info) - .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("hyper", log::LevelFilter::Info) + .level_for("hyper_rustls", log::LevelFilter::Info) + .level_for("reqwest", log::LevelFilter::Info) + .level_for("sqlx", log::LevelFilter::Warn) + .level_for("tao", log::LevelFilter::Info) + .level_for("tokio_util", log::LevelFilter::Info) .level_for("tonic", log::LevelFilter::Info) + .level_for("tower", log::LevelFilter::Info) + .level_for("tracing", log::LevelFilter::Info) .with_colors(ColoredLevelConfig::default()) .level(log::LevelFilter::Trace) .build(), ) - .plugin(tauri_plugin_window_state::Builder::default().build()) .setup(|app| { let app_data_dir = app.path_resolver().app_data_dir().unwrap(); let app_config_dir = app.path_resolver().app_config_dir().unwrap(); @@ -1292,7 +1318,7 @@ fn emit_and_return( } /// Emit an event to all windows, used for side-effects where there is no source window to attribute. This -fn emit_side_effect(app_handle: &AppHandle, event: &str, payload: S) { +fn emit_side_effect(app_handle: AppHandle, event: &str, payload: S) { app_handle.emit_all(event, &payload).unwrap(); } diff --git a/src-web/components/GrpcConnectionLayout.tsx b/src-web/components/GrpcConnectionLayout.tsx index 9685ae97..fa09cc1a 100644 --- a/src-web/components/GrpcConnectionLayout.tsx +++ b/src-web/components/GrpcConnectionLayout.tsx @@ -193,7 +193,7 @@ export function GrpcConnectionLayout({ style }: Props) { + { + await grpc.cancel.mutateAsync(); + }} + icon="trash" + /> { - await invoke('cmd_grpc_cancel', { id: activeConnectionId }); + await emit('grpc_message_in', 'Cancel'); setActiveConnectionId(null); }, }); @@ -92,7 +93,6 @@ export function useGrpc(url: string | null) { queryKey: ['grpc_reflect', url ?? ''], queryFn: async () => { if (url === null) return []; - console.log('GETTING SCHEMA', url); return (await invoke('cmd_grpc_reflect', { endpoint: url })) as ReflectResponseService[]; }, });