diff --git a/crates/lox-distributor/src/main.rs b/crates/lox-distributor/src/main.rs index cf879d8..6998e04 100644 --- a/crates/lox-distributor/src/main.rs +++ b/crates/lox-distributor/src/main.rs @@ -18,7 +18,7 @@ use lox::{ }; use lox::{BridgeAuth, BridgeDb, OPENINV_LENGTH}; use rand::RngCore; -use rdsys_backend::{proto::ResourceDiff, start_stream, ResourceStream}; +use rdsys_backend::{proto::ResourceDiff, start_stream}; use serde::{Deserialize, Serialize}; use std::{ convert::Infallible, @@ -29,13 +29,12 @@ use std::{ sync::{Arc, Mutex}, time::Duration, }; - use serde_json; use serde_with::serde_as; use tokio::{ spawn, - sync::{mpsc, oneshot}, - time::sleep, task::yield_now, + sync::{broadcast, mpsc, oneshot}, + time::sleep, signal, }; #[serde_as] @@ -373,9 +372,21 @@ async fn shutdown_signal() { tokio::signal::ctrl_c() .await .expect("failed to listen for ctrl+c signal"); + println!("Shut down Lox Server"); } -async fn rdsys_sender(rstream: ResourceStream, tx: Sender) { +async fn rdsys_stream(rtype: ResourceInfo, tx: Sender, mut kill: broadcast::Receiver<()>) { + tokio:: select! { + start_rdsys_stream = rdsys_sender(rtype, tx) => start_rdsys_stream , + _ = kill.recv() => {println!("Shut down rdsys stream"); return}, + } +} + +async fn rdsys_sender(rtype: ResourceInfo, tx: Sender) { + let rstream = start_stream(rtype.endpoint, rtype.name, rtype.token, rtype.types) + .await + .expect("rdsys stream initialization failed. Start rdsys or check config.json"); + sleep(Duration::from_millis(1)).await; for diff in rstream { println!("Received diff: {:?}", diff); //send this through a channel tx.send(diff).await.unwrap(); @@ -383,6 +394,13 @@ async fn rdsys_sender(rstream: ResourceStream, tx: Sender) { } } +async fn rdsys_bridge_parser(rdsys_tx: mpsc::Sender, rx: Receiver, mut kill: broadcast::Receiver<()>) { + tokio:: select! { + start_bridge_parser = parse_bridges(rdsys_tx, rx) => start_bridge_parser , + _ = kill.recv() => {println!("Shut down bridge_parser"); return}, + } +} + async fn parse_bridges(rdsys_tx: mpsc::Sender, rx: Receiver) { loop { let resourcediff = rx.recv().await.unwrap(); @@ -394,6 +412,131 @@ async fn parse_bridges(rdsys_tx: mpsc::Sender, rx: Receiver, mut kill: broadcast::Receiver<()>) { + tokio:: select! { + create_context = context_manager(context_rx) => create_context, + _ = kill.recv() => {println!("Shut down context_manager"); return}, + } +} + +async fn context_manager(mut context_rx: mpsc::Receiver) { + // pass in distribution of open invite vs. hot spare buckets? + + let bridgedb = BridgeDb::new(); + let lox_auth = BridgeAuth::new(bridgedb.pubkey); + + let context = LoxServerContext { + db: Arc::new(Mutex::new(bridgedb)), + ba: Arc::new(Mutex::new(lox_auth)), + }; + + while let Some(cmd) = context_rx.recv().await { + use Command::*; + + match cmd { + Rdsys { resourcediff } => { + for new_resource in resourcediff.new { + for pt in new_resource { + println!("A NEW RESOURCE: {:?}", pt); + let mut bucket = [ + BridgeLine::default(), + BridgeLine::default(), + BridgeLine::default(), + ]; + let mut count = 0; + for resource in pt.1 { + let mut ip_bytes: [u8; 16] = [0; 16]; + ip_bytes[..resource.address.len()] + .copy_from_slice(resource.address.as_bytes()); + let infostr: String = format!( + "type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}", + resource.r#type, + resource.blocked_in, + resource.protocol, + resource.fingerprint, + resource.or_addresses, + resource.distribution, + resource.flags, + resource.params, + ); + let mut info_bytes: [u8; BRIDGE_BYTES - 18] = + [0; BRIDGE_BYTES - 18]; + + info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes()); + let bridgeline = BridgeLine { + addr: ip_bytes, + port: resource.port, + info: info_bytes, + }; + + println!("Now it's a bridgeline: {:?}", bridgeline); + if count < 2 { + bucket[count] = bridgeline; + count += 1; + } else { + context.add_openinv_bucket(bucket); + count = 0; + bucket = [ + BridgeLine::default(), + BridgeLine::default(), + BridgeLine::default(), + ]; + } + } + } + } + for changed_resource in resourcediff.changed { + println!("A NEW CHANGED RESOURCE: {:?}", changed_resource); + } + for gone_resource in resourcediff.gone { + for pt in gone_resource { + println!("A NEW GONE RESOURCE: {:?}", pt); + for resource in pt.1 { + let mut ip_bytes: [u8; 16] = [0; 16]; + ip_bytes[..resource.address.len()] + .copy_from_slice(resource.address.as_bytes()); + let infostr: String = format!( + "type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}", + resource.r#type, + resource.blocked_in, + resource.protocol, + resource.fingerprint, + resource.or_addresses, + resource.distribution, + resource.flags, + resource.params, + ); + let mut info_bytes: [u8; BRIDGE_BYTES - 18] = + [0; BRIDGE_BYTES - 18]; + + info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes()); + let bridgeline = BridgeLine { + addr: ip_bytes, + port: resource.port, + info: info_bytes, + }; + + println!("Now it's a bridgeline: {:?}", bridgeline); + context.add_unreachable(bridgeline); + } + } + } + context.encrypt_table(); + sleep(Duration::from_millis(1)).await; + } + Request { req, sender } => { + let response = handle(context.clone(), req).await; + if let Err(e) = sender.send(response) { + eprintln!("Server Response Error: {:?}", e); + }; + sleep(Duration::from_millis(1)).await; + } + } + } + + +} + #[derive(Debug)] enum Command { Rdsys { @@ -405,7 +548,6 @@ enum Command { }, } -// Run with cargo run -- config.json #[tokio::main] async fn main() { let args: Vec = env::args().collect(); @@ -414,133 +556,38 @@ async fn main() { // Read the JSON contents of the file as a ResourceInfo let rtype: ResourceInfo = serde_json::from_reader(reader).unwrap(); - let (rdsys_tx, mut context_rx) = mpsc::channel(32); + let (rdsys_tx, context_rx) = mpsc::channel(32); let request_tx = rdsys_tx.clone(); - let context_manager = spawn(async move { - // pass in distribution of open invite vs. hot spare buckets? - let bridgedb = BridgeDb::new(); - let lox_auth = BridgeAuth::new(bridgedb.pubkey); - let context = LoxServerContext { - db: Arc::new(Mutex::new(bridgedb)), - ba: Arc::new(Mutex::new(lox_auth)), - }; - while let Some(cmd) = context_rx.recv().await { - use Command::*; + // create the shutdown broadcast channel + let (shutdown_tx, mut shutdown_rx) = broadcast::channel(16); + let kill_stream = shutdown_tx.subscribe(); + let kill_parser = shutdown_tx.subscribe(); + let kill_context= shutdown_tx.subscribe(); - match cmd { - Rdsys { resourcediff } => { - for new_resource in resourcediff.new { - for pt in new_resource { - println!("A NEW RESOURCE: {:?}", pt); - let mut bucket = [ - BridgeLine::default(), - BridgeLine::default(), - BridgeLine::default(), - ]; - let mut count = 0; - for resource in pt.1 { - let mut ip_bytes: [u8; 16] = [0; 16]; - ip_bytes[..resource.address.len()] - .copy_from_slice(resource.address.as_bytes()); - let infostr: String = format!( - "type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}", - resource.r#type, - resource.blocked_in, - resource.protocol, - resource.fingerprint, - resource.or_addresses, - resource.distribution, - resource.flags, - resource.params, - ); - let mut info_bytes: [u8; BRIDGE_BYTES - 18] = - [0; BRIDGE_BYTES - 18]; +// Listen for ctrl_c, send signal to broadcast shutdown to all threads by dropping shutdown_tx + let shutdown_handler = spawn(async move { + tokio::select! { + _ = signal::ctrl_c() => { + drop(shutdown_tx); + println!("Kill Sent, all threads should shutdown."); - info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes()); - let bridgeline = BridgeLine { - addr: ip_bytes, - port: resource.port, - info: info_bytes, - }; - - println!("Now it's a bridgeline: {:?}", bridgeline); - if count < 2 { - bucket[count] = bridgeline; - count += 1; - } else { - context.add_openinv_bucket(bucket); - count = 0; - bucket = [ - BridgeLine::default(), - BridgeLine::default(), - BridgeLine::default(), - ]; - } - } - } - } - for changed_resource in resourcediff.changed { - println!("A NEW CHANGED RESOURCE: {:?}", changed_resource); - } - for gone_resource in resourcediff.gone { - for pt in gone_resource { - println!("A NEW GONE RESOURCE: {:?}", pt); - for resource in pt.1 { - let mut ip_bytes: [u8; 16] = [0; 16]; - ip_bytes[..resource.address.len()] - .copy_from_slice(resource.address.as_bytes()); - let infostr: String = format!( - "type={} blocked_in={:?} protocol={} fingerprint={} or_addresses={:?} distribution={} flags={:?} params={:?}", - resource.r#type, - resource.blocked_in, - resource.protocol, - resource.fingerprint, - resource.or_addresses, - resource.distribution, - resource.flags, - resource.params, - ); - let mut info_bytes: [u8; BRIDGE_BYTES - 18] = - [0; BRIDGE_BYTES - 18]; - - info_bytes[..infostr.len()].copy_from_slice(infostr.as_bytes()); - let bridgeline = BridgeLine { - addr: ip_bytes, - port: resource.port, - info: info_bytes, - }; - - println!("Now it's a bridgeline: {:?}", bridgeline); - context.add_unreachable(bridgeline); - } - } - } - context.encrypt_table(); - sleep(Duration::from_millis(1)).await; - } - Request { req, sender } => { - let response = handle(context.clone(), req).await; - if let Err(e) = sender.send(response) { - eprintln!("Server Response Error: {:?}", e); - }; - sleep(Duration::from_millis(1)).await; - } + _ = shutdown_rx.recv().await; + println!("Receiving shutdown signals. . ."); } } - }); + }); + + + + let context_manager = spawn(async move { create_context_manager(context_rx, kill_context).await }); - //Sender is resource stream and receiver is bridgedb function (add_openinv_bridges) let (tx, rx) = async_channel::unbounded(); - // to populate the bridge db - let rstream = start_stream(rtype.endpoint, rtype.name, rtype.token, rtype.types) - .await - .unwrap(); - let rdsys_stream_handler = spawn(async { rdsys_sender(rstream, tx).await }); + let rdsys_stream_handler = spawn(async { rdsys_stream(rtype, tx, kill_stream).await }); - let rdsys_resource_receiver = spawn(async { parse_bridges(rdsys_tx, rx).await }); + let rdsys_resource_receiver = spawn(async { rdsys_bridge_parser(rdsys_tx, rx, kill_parser).await }); let make_service = make_service_fn(move |_conn: &AddrStream| { let request_tx = request_tx.clone(); @@ -561,11 +608,18 @@ 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, context_manager]).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, + shutdown_handler, + ]) + .await; + }