diff --git a/crates/lox-distributor/src/main.rs b/crates/lox-distributor/src/main.rs index cff7343..cf879d8 100644 --- a/crates/lox-distributor/src/main.rs +++ b/crates/lox-distributor/src/main.rs @@ -4,7 +4,6 @@ use hyper::{ body, body::Bytes, header::HeaderValue, - http::response, server::conn::AddrStream, service::{make_service_fn, service_fn}, Body, Method, Request, Response, Server, StatusCode, @@ -33,7 +32,11 @@ use std::{ use serde_json; use serde_with::serde_as; -use tokio::{spawn, sync::{mpsc, oneshot}, time::sleep}; +use tokio::{ + spawn, + sync::{mpsc, oneshot}, + time::sleep, task::yield_now, +}; #[serde_as] #[derive(Serialize, Deserialize)] @@ -209,7 +212,6 @@ impl LoxServerContext { } } - async fn handle( cloned_context: LoxServerContext, req: Request, @@ -225,7 +227,9 @@ async fn handle( .unwrap()), _ => match (req.method(), req.uri().path()) { (&Method::GET, "/invite") => Ok::<_, Infallible>(generate_invite(cloned_context)), - (&Method::GET, "/reachability") => Ok::<_, Infallible>(send_reachability_cred(cloned_context)), + (&Method::GET, "/reachability") => { + Ok::<_, Infallible>(send_reachability_cred(cloned_context)) + } (&Method::GET, "/pubkeys") => Ok::<_, Infallible>(send_keys(cloned_context)), (&Method::POST, "/openreq") => Ok::<_, Infallible>({ let bytes = body::to_bytes(req.into_body()).await.unwrap(); @@ -386,16 +390,19 @@ async fn parse_bridges(rdsys_tx: mpsc::Sender, rx: Receiver, - sender: oneshot::Sender, Infallible>>, - }, + sender: oneshot::Sender, Infallible>>, + }, } // Run with cargo run -- config.json @@ -408,7 +415,7 @@ async fn main() { let rtype: ResourceInfo = serde_json::from_reader(reader).unwrap(); let (rdsys_tx, mut context_rx) = mpsc::channel(32); - let mut request_tx = rdsys_tx.clone(); + let request_tx = rdsys_tx.clone(); let context_manager = spawn(async move { // pass in distribution of open invite vs. hot spare buckets? @@ -512,17 +519,21 @@ async fn main() { } } context.encrypt_table(); + sleep(Duration::from_millis(1)).await; } Request { req, sender } => { let response = handle(context.clone(), req).await; - sender.send(response); + if let Err(e) = sender.send(response) { + eprintln!("Server Response Error: {:?}", e); + }; + sleep(Duration::from_millis(1)).await; } } } }); //Sender is resource stream and receiver is bridgedb function (add_openinv_bridges) - let (tx, mut rx) = async_channel::unbounded(); + let (tx, rx) = async_channel::unbounded(); // to populate the bridge db let rstream = start_stream(rtype.endpoint, rtype.name, rtype.token, rtype.types) .await @@ -531,10 +542,9 @@ async fn main() { let rdsys_resource_receiver = spawn(async { parse_bridges(rdsys_tx, rx).await }); - - let make_service = make_service_fn(|_conn: &AddrStream| { + let make_service = make_service_fn(move |_conn: &AddrStream| { let request_tx = request_tx.clone(); - let service = service_fn( move |req| { + let service = service_fn(move |req| { let request_tx = request_tx.clone(); let (response_tx, response_rx) = oneshot::channel(); let cmd = Command::Request { @@ -542,8 +552,8 @@ async fn main() { sender: response_tx, }; async move { - request_tx.send(cmd).await.unwrap(); - response_rx.await.unwrap() + request_tx.send(cmd).await.unwrap(); + response_rx.await.unwrap() } }); async move { Ok::<_, Infallible>(service) } @@ -551,14 +561,11 @@ async fn main() { let addr = SocketAddr::from(([127, 0, 0, 1], 8001)); let server = Server::bind(&addr).serve(make_service); - let graceful = server.with_graceful_shutdown(shutdown_signal()); - println!("Listening on {}", addr); - if let Err(e) = graceful.await { - eprintln!("server error: {}", e); - } - - future::join_all([rdsys_stream_handler, rdsys_resource_receiver]).await; + let graceful = server.with_graceful_shutdown(shutdown_signal()); + println!("Listening on {}", addr); + if let Err(e) = graceful.await { + eprintln!("server error: {}", e); + } + future::join_all([rdsys_stream_handler, rdsys_resource_receiver, context_manager]).await; } - -