starred/yaak
1
0
Fork 0
mirror of https://github.com/mountain-loop/yaak.git synced 2026-03-25 19:01:38 +01:00
Code Issues 8 Packages Projects Releases 10 Wiki Activity

Better connection management

Browse Source
This commit is contained in:
Gregory Schier
2024-02-01 20:29:32 -08:00
parent e3a2b7146b
commit b526ea506b
8 changed files with 173 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
src-tauri/grpc/src/lib.rs
View File

@@ -10,8 +10,8 @@ use crate::proto::{fill_pool, method_desc_to_path};
mod codec;
mod json_schema;
mod proto;
pub mod manager;
mod proto;
pub fn serialize_options() -> SerializeOptions {
SerializeOptions::new().skip_default_fields(false)
@@ -64,7 +64,7 @@ pub async fn client_streaming(
DynamicMessage::deserialize(input_message, &mut deserializer).map_err(|e| e.to_string())?;
deserializer.end().unwrap();
let mut client = tonic::client::Grpc::new(conn);
let mut client = tonic::client::Grpc::with_origin(conn, uri.clone());
println!(
"\n---------- SENDING -----------------\n{}",
@@ -84,43 +84,6 @@ pub async fn client_streaming(
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> {
let (pool, _) = fill_pool(uri).await;

29
src-tauri/grpc/src/manager.rs
View File

@@ -1,10 +1,13 @@
use std::collections::HashMap;
use hyper::client::HttpConnector;
use hyper::Client;
use hyper_rustls::HttpsConnector;
use prost_reflect::{DescriptorPool, DynamicMessage};
use serde_json::Deserializer;
use tokio::sync::mpsc;
use tokio_stream::StreamExt;
use tonic::transport::{Channel, Uri};
use tonic::body::BoxBody;
use tonic::transport::Uri;
use tonic::{IntoRequest, Streaming};
use crate::codec::DynamicCodec;
@@ -15,7 +18,8 @@ type Result<T> = std::result::Result<T, String>;
#[derive(Clone)]
pub struct GrpcConnection {
pool: DescriptorPool,
conn: Channel,
conn: Client<HttpsConnector<HttpConnector>, BoxBody>,
pub uri: Uri,
}
impl GrpcConnection {
@@ -29,7 +33,7 @@ impl GrpcConnection {
.map_err(|e| e.to_string())?;
deserializer.end().unwrap();
let mut client = tonic::client::Grpc::new(self.conn.clone());
let mut client = tonic::client::Grpc::with_origin(self.conn.clone(), self.uri.clone());
println!(
"\n---------- SENDING -----------------\n{}",
@@ -63,7 +67,7 @@ impl GrpcConnection {
.map_err(|e| e.to_string())?;
deserializer.end().unwrap();
let mut client = tonic::client::Grpc::new(self.conn.clone());
let mut client = tonic::client::Grpc::with_origin(self.conn.clone(), self.uri.clone());
println!(
"\n---------- SENDING -----------------\n{}",
@@ -115,24 +119,11 @@ impl GrpcManager {
.await
.server_streaming(service, method, message)
.await
// while let Some(item) = stream.next().await {
// match item {
// Ok(item) => {
// let item = serde_json::to_string_pretty(&item).unwrap();
// println!("Sending message {}", item);
// self.send.send(item).await.unwrap()
// }
// Err(e) => println!("\terror: {}", e),
// }
// }
// Ok(())
}
pub async fn connect(&mut self, id: &str, uri: Uri) -> GrpcConnection {
let (pool, conn) = fill_pool(&uri).await;
let connection = GrpcConnection { pool, conn };
let connection = GrpcConnection { pool, conn, uri };
self.connections.insert(id.to_string(), connection.clone());
connection
}

51
src-tauri/grpc/src/proto.rs
View File

@@ -1,27 +1,48 @@
use std::ops::Deref;
use std::str::FromStr;
use anyhow::anyhow;
use hyper::client::HttpConnector;
use hyper::Client;
use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder};
use prost::Message;
use prost_reflect::{DescriptorPool, MethodDescriptor};
use prost_types::FileDescriptorProto;
use tokio_stream::StreamExt;
use tonic::body::BoxBody;
use tonic::codegen::http::uri::PathAndQuery;
use tonic::transport::Uri;
use tonic::Request;
use tonic::transport::{Channel, Uri};
use tonic_reflection::pb::server_reflection_client::ServerReflectionClient;
use tonic_reflection::pb::server_reflection_request::MessageRequest;
use tonic_reflection::pb::server_reflection_response::MessageResponse;
use tonic_reflection::pb::ServerReflectionRequest;
pub async fn fill_pool(uri: &Uri) -> (DescriptorPool, Channel) {
pub async fn fill_pool(
uri: &Uri,
) -> (
DescriptorPool,
Client<HttpsConnector<HttpConnector>, BoxBody>,
) {
let mut pool = DescriptorPool::new();
let conn = tonic::transport::Endpoint::new(uri.clone())
.unwrap()
.connect()
.await
.unwrap();
let connector = HttpsConnectorBuilder::new().with_native_roots();
let connector = connector.https_or_http().enable_http2().wrap_connector({
let mut http_connector = HttpConnector::new();
http_connector.enforce_http(false);
http_connector
});
let transport = Client::builder()
.pool_max_idle_per_host(0)
.http2_only(true)
.build(connector);
let mut client = ServerReflectionClient::new(conn.clone());
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;
for service in services {
@@ -31,10 +52,12 @@ pub async fn fill_pool(uri: &Uri) -> (DescriptorPool, Channel) {
file_descriptor_set_from_service_name(&service, &mut pool, &mut client).await;
}
(pool, conn)
(pool, transport)
}
async fn list_services(reflect_client: &mut ServerReflectionClient<Channel>) -> Vec<String> {
async fn list_services(
reflect_client: &mut ServerReflectionClient<Client<HttpsConnector<HttpConnector>, BoxBody>>,
) -> Vec<String> {
let response =
send_reflection_request(reflect_client, MessageRequest::ListServices("".into())).await;
@@ -53,13 +76,13 @@ async fn list_services(reflect_client: &mut ServerReflectionClient<Channel>) ->
async fn file_descriptor_set_from_service_name(
service_name: &str,
pool: &mut DescriptorPool,
client: &mut ServerReflectionClient<Channel>,
client: &mut ServerReflectionClient<Client<HttpsConnector<HttpConnector>, BoxBody>>,
) {
let response = send_reflection_request(
client,
MessageRequest::FileContainingSymbol(service_name.into()),
)
.await;
.await;
let file_descriptor_response = match response {
MessageResponse::FileDescriptorResponse(resp) => resp,
@@ -82,7 +105,7 @@ async fn file_descriptor_set_from_service_name(
async fn file_descriptor_set_by_filename(
filename: &str,
pool: &mut DescriptorPool,
client: &mut ServerReflectionClient<Channel>,
client: &mut ServerReflectionClient<Client<HttpsConnector<HttpConnector>, BoxBody>>,
) {
// We already fetched this file
if let Some(_) = pool.get_file_by_name(filename) {
@@ -104,7 +127,7 @@ async fn file_descriptor_set_by_filename(
}
async fn send_reflection_request(
client: &mut ServerReflectionClient<Channel>,
client: &mut ServerReflectionClient<Client<HttpsConnector<HttpConnector>, BoxBody>>,
message: MessageRequest,
) -> MessageResponse {
let reflection_request = ServerReflectionRequest {