diff --git a/src-tauri/grpc/src/manager.rs b/src-tauri/grpc/src/manager.rs index e6c8fce6..f51b6989 100644 --- a/src-tauri/grpc/src/manager.rs +++ b/src-tauri/grpc/src/manager.rs @@ -4,8 +4,8 @@ use std::path::PathBuf; use hyper::client::HttpConnector; use hyper::Client; use hyper_rustls::HttpsConnector; -use prost_reflect::DescriptorPool; pub use prost_reflect::DynamicMessage; +use prost_reflect::{DescriptorPool, MethodDescriptor, ServiceDescriptor}; use serde_json::Deserializer; use tokio_stream::wrappers::ReceiverStream; use tokio_stream::StreamExt; @@ -25,14 +25,30 @@ pub struct GrpcConnection { } impl GrpcConnection { + pub fn service(&self, service: &str) -> Result { + let service = self + .pool + .get_service_by_name(service) + .ok_or("Failed to find service")?; + Ok(service) + } + + pub fn method(&self, service: &str, method: &str) -> Result { + let service = self.service(service)?; + let method = service + .methods() + .find(|m| m.name() == method) + .ok_or("Failed to find method")?; + Ok(method) + } + pub async fn unary( &self, service: &str, method: &str, message: &str, ) -> Result { - let service = self.pool.get_service_by_name(service).unwrap(); - let method = &service.methods().find(|m| m.name() == method).unwrap(); + let method = &self.method(&service, &method)?; let input_message = method.input(); let mut deserializer = Deserializer::from_str(message); @@ -60,9 +76,7 @@ impl GrpcConnection { method: &str, stream: ReceiverStream, ) -> Result, String> { - let service = self.pool.get_service_by_name(service).unwrap(); - let method = &service.methods().find(|m| m.name() == method).unwrap(); - + let method = &self.method(&service, &method)?; let mut client = tonic::client::Grpc::with_origin(self.conn.clone(), self.uri.clone()); let method2 = method.clone(); @@ -92,8 +106,7 @@ impl GrpcConnection { method: &str, stream: ReceiverStream, ) -> Result { - let service = self.pool.get_service_by_name(service).unwrap(); - let method = &service.methods().find(|m| m.name() == method).unwrap(); + let method = &self.method(&service, &method)?; let mut client = tonic::client::Grpc::with_origin(self.conn.clone(), self.uri.clone()); let req = { @@ -126,8 +139,7 @@ impl GrpcConnection { method: &str, message: &str, ) -> Result, String> { - let service = self.pool.get_service_by_name(service).unwrap(); - let method = &service.methods().find(|m| m.name() == method).unwrap(); + let method = &self.method(&service, &method)?; let input_message = method.input(); let mut deserializer = Deserializer::from_str(message); diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index 26b3c0c5..2b845a59 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -127,100 +127,11 @@ async fn cmd_grpc_reflect( } #[tauri::command] -async fn cmd_grpc_call_unary( +async fn cmd_grpc_go( request_id: &str, w: Window, grpc_handle: State<'_, Mutex>, -) -> Result { - let req = get_grpc_request(&w, request_id) - .await - .map_err(|e| e.to_string())?; - let conn = { - let req = req.clone(); - upsert_grpc_connection( - &w, - &GrpcConnection { - workspace_id: req.workspace_id, - request_id: req.id, - ..Default::default() - }, - ) - .await - .map_err(|e| e.to_string())? - }; - - { - let req = req.clone(); - let conn = conn.clone(); - upsert_grpc_message( - &w, - &GrpcMessage { - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn.id, - is_info: true, - message: format!("Initiating connection to {}", req.url), - ..Default::default() - }, - ) - .await - .map_err(|e| e.to_string())?; - }; - - let uri = safe_uri(&req.url).map_err(|e| e.to_string())?; - let start = std::time::Instant::now(); - let msg = match grpc_handle - .lock() - .await - .connect( - &req.clone().id, - uri, - req.proto_files - .0 - .iter() - .map(|p| PathBuf::from_str(p).unwrap()) - .collect(), - ) - .await? - .unary( - &req.service.unwrap_or_default(), - &req.method.unwrap_or_default(), - &req.message, - ) - .await - { - Ok(msg) => { - upsert_grpc_message( - &w, - &GrpcMessage { - message: serialize_message(&msg)?, - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn.clone().id, - is_server: true, - ..Default::default() - }, - ) - .await - } - Err(e) => return Err(e.to_string()), - }; - - upsert_grpc_connection( - &w, - &GrpcConnection { - elapsed: start.elapsed().as_millis() as i64, - ..conn - }, - ) - .await - .map_err(|e| e.to_string())?; - - msg.map_err(|e| e.to_string()) -} - -#[tauri::command] -async fn cmd_grpc_client_streaming(request_id: &str, w: Window) -> Result { +) -> Result { let req = get_grpc_request(&w, request_id) .await .map_err(|e| e.to_string())?; @@ -253,7 +164,7 @@ async fn cmd_grpc_client_streaming(request_id: &str, w: Window) -> Result(16); @@ -272,11 +183,30 @@ async fn cmd_grpc_client_streaming(request_id: &str, w: Window) -> Result Result Result>(); - let msg = grpc_handle - .lock() - .await - .connect( - &req.clone().id, - uri, - req.proto_files - .0 - .iter() - .map(|p| PathBuf::from_str(p).unwrap()) - .collect(), + let (maybe_stream, maybe_msg) = match ( + method_desc.is_client_streaming(), + method_desc.is_server_streaming(), + ) { + (true, true) => ( + Some( + connection + .streaming(&service, &method, in_msg_stream) + .await + .unwrap(), + ), + None, + ), + (true, false) => ( + None, + Some( + connection + .client_streaming(&service, &method, in_msg_stream) + .await + .map_err(|e| e.to_string()) + .unwrap(), + ), + ), + (false, true) => ( + Some( + connection + .server_streaming(&service, &method, &req.message) + .await + .unwrap(), + ), + None, + ), + (false, false) => ( + None, + Some( + connection + .unary(&service, &method, &req.message) + .await + .unwrap(), + ), + ), + }; + + if let Some(msg) = maybe_msg { + let req = req.clone(); + let conn = conn.clone(); + upsert_grpc_message( + &w, + &GrpcMessage { + message: serialize_message(&msg).unwrap(), + workspace_id: req.workspace_id, + request_id: req.id, + connection_id: conn.id, + is_server: true, + ..Default::default() + }, ) .await - .unwrap() - .client_streaming(&service, &method, in_msg_stream) - .await + .map_err(|e| e.to_string()) .unwrap(); - let message = serialize_message(&msg).unwrap(); - upsert_grpc_message( - &w, - &GrpcMessage { - message, - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn.id, - is_server: true, - ..Default::default() - }, - ) - .await - .unwrap(); - } - }; - - { - let conn = conn.clone(); - let w = w.clone(); - tauri::async_runtime::spawn(async move { - tokio::select! { - _ = grpc_listen => { - upsert_grpc_message( - &w, - &GrpcMessage { - message: "Connection completed".to_string(), - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn.clone().id, - is_info: true, - ..Default::default() - }, - ) - .await.unwrap(); - upsert_grpc_connection( - &w, - &GrpcConnection { - elapsed: start.elapsed().as_millis() as i64, - ..conn - }, - ) - .await - .unwrap(); - }, - _ = cancelled_rx.changed() => { - upsert_grpc_message( - &w, - &GrpcMessage { - message: "Connection cancelled".to_string(), - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn.clone().id, - is_info: true, - ..Default::default() - }, - ) - .await.unwrap(); - - upsert_grpc_connection( - &w, - &GrpcConnection { - elapsed: start.elapsed().as_millis() as i64, - ..conn - }, - ) - .await - .unwrap(); - }, } - w.unlisten(event_handler); - }); - }; - - Ok(conn) -} - -#[tauri::command] -async fn cmd_grpc_streaming( - request_id: &str, - w: Window, - grpc_handle: State<'_, Mutex>, -) -> Result { - let req = get_grpc_request(&w, request_id) - .await - .map_err(|e| e.to_string())?; - let conn = { - let req = req.clone(); - upsert_grpc_connection( - &w, - &GrpcConnection { - workspace_id: req.workspace_id, - request_id: req.id, - ..Default::default() - }, - ) - .await - .map_err(|e| e.to_string())? - }; - - { - let conn = conn.clone(); - let req = req.clone(); - upsert_grpc_message( - &w, - &GrpcMessage { - message: "Initiating connection".to_string(), - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn.id, - is_info: true, - ..Default::default() - }, - ) - .await - .expect("Failed to upsert message"); - }; - - let (in_msg_tx, in_msg_rx) = tauri::async_runtime::channel::(16); - let (cancelled_tx, mut cancelled_rx) = tokio::sync::watch::channel(false); - - let uri = safe_uri(&req.url).map_err(|e| e.to_string())?; - - let in_msg_stream = tokio_stream::wrappers::ReceiverStream::new(in_msg_rx); - - let (service, method) = { - let req = req.clone(); - match (req.service, req.method) { - (Some(service), Some(method)) => (service, method), - _ => return Err("Service and method are required".to_string()), - } - }; - - let start = std::time::Instant::now(); - let mut stream = grpc_handle - .lock() - .await - .connect( - &req.clone().id, - uri, - req.proto_files - .0 - .iter() - .map(|p| PathBuf::from_str(p).unwrap()) - .collect(), - ) - .await? - .streaming(&service, &method, in_msg_stream) - .await - .unwrap(); - - #[derive(serde::Deserialize)] - enum IncomingMsg { - Message(String), - Cancel, - } - - let cb = { - let cancelled_rx = cancelled_rx.clone(); - let w = w.clone(); - let conn = conn.clone(); - let req = req.clone(); - - move |ev: tauri::Event| { - if *cancelled_rx.borrow() { - // Stream is cancelled - return; - } - - match serde_json::from_str::(ev.payload().unwrap()) { - Ok(IncomingMsg::Message(msg)) => { - in_msg_tx.try_send(msg.clone()).unwrap(); - let w = w.clone(); - let req = req.clone(); - let conn = conn.clone(); - tauri::async_runtime::spawn(async move { - upsert_grpc_message( - &w, - &GrpcMessage { - message: msg, - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn.id, - ..Default::default() - }, - ) - .await - .map_err(|e| e.to_string()) - .unwrap(); - }); - } - Ok(IncomingMsg::Cancel) => { - cancelled_tx.send_replace(true); - } - Err(e) => { - error!("Failed to parse gRPC message: {:?}", e); - } - } - } - }; - let event_handler = w.listen_global(format!("grpc_client_msg_{}", conn.id).as_str(), cb); - - let grpc_listen = { - let w = w.clone(); - let conn = conn.clone(); - let req = req.clone(); - async move { + let mut stream = match maybe_stream { + Some(s) => s, + None => return, + }; loop { match stream.next().await { Some(Ok(item)) => { @@ -605,6 +375,7 @@ async fn cmd_grpc_streaming( let conn = conn.clone(); tauri::async_runtime::spawn(async move { let w = w.clone(); + let req = req.clone(); tokio::select! { _ = grpc_listen => { upsert_grpc_message( @@ -656,196 +427,6 @@ async fn cmd_grpc_streaming( Ok(conn.id) } -#[tauri::command] -async fn cmd_grpc_server_streaming( - request_id: &str, - w: Window, - grpc_handle: State<'_, Mutex>, -) -> Result { - let req = get_grpc_request(&w, request_id) - .await - .map_err(|e| e.to_string())?; - - let conn = { - let req = req.clone(); - upsert_grpc_connection( - &w, - &GrpcConnection { - workspace_id: req.workspace_id, - request_id: req.id, - ..Default::default() - }, - ) - .await - .map_err(|e| e.to_string())? - }; - - { - let req = req.clone(); - let conn = conn.clone(); - upsert_grpc_message( - &w, - &GrpcMessage { - message: "Initiating connection".to_string(), - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn.id, - is_info: true, - ..Default::default() - }, - ) - .await - .unwrap(); - } - - let (cancelled_tx, mut cancelled_rx) = tokio::sync::watch::channel(false); - - let (service, method) = match (&req.service, &req.method) { - (Some(service), Some(method)) => (service, method), - _ => return Err("Service and method are required".to_string()), - }; - - let uri = safe_uri(&req.url).map_err(|e| e.to_string())?; - let mut stream = grpc_handle - .lock() - .await - .connect( - &req.clone().id, - uri, - req.proto_files - .0 - .iter() - .map(|p| PathBuf::from_str(p).unwrap()) - .collect(), - ) - .await? - .server_streaming(&service, &method, &req.message) - .await - .expect("FAILED"); - - #[derive(serde::Deserialize)] - enum IncomingMsg { - 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::(ev.payload().unwrap()) { - Ok(IncomingMsg::Cancel) => { - cancelled_tx.send_replace(true); - } - Err(e) => { - error!("Failed to parse gRPC message: {:?}", e); - } - } - } - }; - let event_handler = w.listen_global(format!("grpc_client_msg_{}", conn.id).as_str(), cb); - - let start = std::time::Instant::now(); - let grpc_listen = { - let conn_id = conn.clone().id; - let w = w.clone(); - let req = req.clone(); - async move { - loop { - let req = req.clone(); - let conn_id = conn_id.clone(); - let w = w.clone(); - match stream.next().await { - Some(Ok(item)) => { - let item = serialize_message(&item).unwrap(); - upsert_grpc_message( - &w, - &GrpcMessage { - message: item, - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn_id, - is_server: true, - ..Default::default() - }, - ) - .await - .map_err(|e| e.to_string()) - .expect("Failed to upsert message"); - } - Some(Err(e)) => { - error!("gRPC stream error: {:?}", e); - // TODO: Handle error - } - None => { - info!("gRPC stream closed by sender"); - break; - } - } - } - } - }; - - { - let conn = conn.clone(); - let req = req.clone(); - let w = w.clone(); - tauri::async_runtime::spawn(async move { - tokio::select! { - _ = grpc_listen => { - upsert_grpc_message( - &w, - &GrpcMessage { - message: "Connection completed".to_string(), - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn.clone().id, - is_info: true, - ..Default::default() - }, - ) - .await.unwrap(); - upsert_grpc_connection( - &w, - &GrpcConnection{ - elapsed: start.elapsed().as_millis() as i64, - ..conn - }, - ).await.unwrap(); - }, - _ = cancelled_rx.changed() => { - upsert_grpc_message( - &w, - &GrpcMessage { - message: "Connection cancelled".to_string(), - workspace_id: req.workspace_id, - request_id: req.id, - connection_id: conn.clone().id, - is_info: true, - ..Default::default() - }, - ) - .await.unwrap(); - upsert_grpc_connection( - &w, - &GrpcConnection{ - elapsed: start.elapsed().as_millis() as i64, - ..conn - }, - ).await.unwrap(); - }, - } - w.unlisten(event_handler); - }); - } - - Ok(conn) -} - #[tauri::command] async fn cmd_send_ephemeral_request( mut request: HttpRequest, @@ -1637,10 +1218,7 @@ fn main() { cmd_get_grpc_request, cmd_get_settings, cmd_get_workspace, - cmd_grpc_call_unary, - cmd_grpc_client_streaming, - cmd_grpc_server_streaming, - cmd_grpc_streaming, + cmd_grpc_go, cmd_grpc_reflect, cmd_import_data, cmd_list_cookie_jars, diff --git a/src-web/components/GlobalHooks.tsx b/src-web/components/GlobalHooks.tsx index 5c5d7a0e..63833273 100644 --- a/src-web/components/GlobalHooks.tsx +++ b/src-web/components/GlobalHooks.tsx @@ -85,8 +85,10 @@ export function GlobalHooks() { queryClient.setQueryData(queryKey, (values = []) => { const index = values.findIndex((v) => modelsEq(v, payload)) ?? -1; if (index >= 0) { + // console.log('UPDATED', payload); return [...values.slice(0, index), payload, ...values.slice(index + 1)]; } else { + // console.log('CREATED', payload); return pushToFront ? [payload, ...(values ?? [])] : [...(values ?? []), payload]; } }); diff --git a/src-web/components/GrpcEditor.tsx b/src-web/components/GrpcEditor.tsx index 12964bf5..adcf4002 100644 --- a/src-web/components/GrpcEditor.tsx +++ b/src-web/components/GrpcEditor.tsx @@ -36,10 +36,18 @@ export function GrpcEditor({ // Find the schema for the selected service and method and update the editor useEffect(() => { - if (editorViewRef.current == null || services === null) return; + if ( + editorViewRef.current == null || + services === null || + request.service === null || + request.method === null + ) { + return; + } const s = services.find((s) => s.name === request.service); - if (request.service != null && s == null) { + if (s == null) { + console.log('Failed to find service', { service: request.service, services }); alert({ id: 'grpc-find-service-error', title: "Couldn't Find Service", @@ -52,8 +60,9 @@ export function GrpcEditor({ return; } - const schema = s?.methods.find((m) => m.name === request.method)?.schema; + const schema = s.methods.find((m) => m.name === request.method)?.schema; if (request.method != null && schema == null) { + console.log('Failed to find method', { method: request.method, methods: s?.methods }); alert({ id: 'grpc-find-schema-error', title: "Couldn't Find Method", diff --git a/src-web/components/Sidebar.tsx b/src-web/components/Sidebar.tsx index da2231f0..bbdee658 100644 --- a/src-web/components/Sidebar.tsx +++ b/src-web/components/Sidebar.tsx @@ -493,7 +493,7 @@ function SidebarItems({ child.item.model === 'http_request' ? ( {child.item.method} ) : child.item.model === 'grpc_request' ? ( - gRPC + GRPC ) : null } onMove={handleMove} diff --git a/src-web/components/core/HttpMethodTag.tsx b/src-web/components/core/HttpMethodTag.tsx index 2e5d51d5..c336ee7b 100644 --- a/src-web/components/core/HttpMethodTag.tsx +++ b/src-web/components/core/HttpMethodTag.tsx @@ -13,7 +13,7 @@ const methodMap: Record = { delete: 'DELETE', options: 'OPTIONS', head: 'HEAD', - grpc: 'gRPC', + grpc: 'GRPC', }; export function HttpMethodTag({ children: method, className }: Props) { diff --git a/src-web/components/core/JsonAttributeTree.tsx b/src-web/components/core/JsonAttributeTree.tsx index 860f06d3..48a80c88 100644 --- a/src-web/components/core/JsonAttributeTree.tsx +++ b/src-web/components/core/JsonAttributeTree.tsx @@ -31,6 +31,7 @@ export const JsonAttributeTree = ({ depth = 0, attrKey, attrValue, attrKeyJsonPa .sort((a, b) => a.localeCompare(b)) .flatMap((k) => ( ( ({ mutationFn: async () => { if (id === null) throw new Error("Can't duplicate a null grpc request"); @@ -25,10 +23,6 @@ export function useDuplicateGrpcRequest({ }, onSettled: () => trackEvent('GrpcRequest', 'Duplicate'), onSuccess: async (request) => { - queryClient.setQueryData( - grpcRequestsQueryKey({ workspaceId: request.workspaceId }), - (requests) => [...(requests ?? []), request], - ); if (navigateAfter && activeWorkspaceId !== null) { routes.navigate('request', { workspaceId: activeWorkspaceId, diff --git a/src-web/hooks/useDuplicateHttpRequest.ts b/src-web/hooks/useDuplicateHttpRequest.ts index 6e1f33f9..a7077d0a 100644 --- a/src-web/hooks/useDuplicateHttpRequest.ts +++ b/src-web/hooks/useDuplicateHttpRequest.ts @@ -1,11 +1,10 @@ -import { useMutation, useQueryClient } from '@tanstack/react-query'; +import { useMutation } from '@tanstack/react-query'; import { invoke } from '@tauri-apps/api'; import { trackEvent } from '../lib/analytics'; import type { HttpRequest } from '../lib/models'; import { useActiveEnvironmentId } from './useActiveEnvironmentId'; import { useActiveWorkspaceId } from './useActiveWorkspaceId'; import { useAppRoutes } from './useAppRoutes'; -import { httpRequestsQueryKey } from './useHttpRequests'; export function useDuplicateHttpRequest({ id, @@ -17,7 +16,6 @@ export function useDuplicateHttpRequest({ const activeWorkspaceId = useActiveWorkspaceId(); const activeEnvironmentId = useActiveEnvironmentId(); const routes = useAppRoutes(); - const queryClient = useQueryClient(); return useMutation({ mutationFn: async () => { if (id === null) throw new Error("Can't duplicate a null request"); @@ -25,10 +23,6 @@ export function useDuplicateHttpRequest({ }, onSettled: () => trackEvent('HttpRequest', 'Duplicate'), onSuccess: async (request) => { - queryClient.setQueryData( - httpRequestsQueryKey({ workspaceId: request.workspaceId }), - (requests) => [...(requests ?? []), request], - ); if (navigateAfter && activeWorkspaceId !== null) { routes.navigate('request', { workspaceId: activeWorkspaceId, diff --git a/src-web/hooks/useGrpc.ts b/src-web/hooks/useGrpc.ts index c04bfd1f..c6d74856 100644 --- a/src-web/hooks/useGrpc.ts +++ b/src-web/hooks/useGrpc.ts @@ -2,7 +2,7 @@ import { useMutation, useQuery } from '@tanstack/react-query'; import { invoke } from '@tauri-apps/api'; import { emit } from '@tauri-apps/api/event'; import { minPromiseMillis } from '../lib/minPromiseMillis'; -import type { GrpcConnection, GrpcMessage, GrpcRequest } from '../lib/models'; +import type { GrpcConnection, GrpcRequest } from '../lib/models'; import { useDebouncedValue } from './useDebouncedValue'; export interface ReflectResponseService { @@ -13,27 +13,20 @@ export interface ReflectResponseService { export function useGrpc(req: GrpcRequest | null, conn: GrpcConnection | null) { const requestId = req?.id ?? 'n/a'; - const unary = useMutation({ - mutationKey: ['grpc_unary', conn?.id ?? 'n/a'], - mutationFn: async () => - (await invoke('cmd_grpc_call_unary', { - requestId, - })) as GrpcMessage, + const unary = useMutation({ + mutationFn: async () => await invoke('cmd_grpc_go', { requestId }), }); const clientStreaming = useMutation({ - mutationKey: ['grpc_client_streaming', conn?.id ?? 'n/a'], - mutationFn: async () => await invoke('cmd_grpc_client_streaming', { requestId }), + mutationFn: async () => await invoke('cmd_grpc_go', { requestId }), }); const serverStreaming = useMutation({ - mutationKey: ['grpc_server_streaming', conn?.id ?? 'n/a'], - mutationFn: async () => await invoke('cmd_grpc_server_streaming', { requestId }), + mutationFn: async () => await invoke('cmd_grpc_go', { requestId }), }); const streaming = useMutation({ - mutationKey: ['grpc_streaming', conn?.id ?? 'n/a'], - mutationFn: async () => await invoke('cmd_grpc_streaming', { requestId }), + mutationFn: async () => await invoke('cmd_grpc_go', { requestId }), }); const send = useMutation({