vecna
/
lox
mirror of https://gitlab.torproject.org/vecna/lox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Server responds to requests and processes bridgelines from rdsys

This commit is contained in:
onyinyang 2023-03-15 15:21:03 -04:00
parent ad6c8278d2
commit 3226cb5a1e
No known key found for this signature in database
GPG Key ID: 156A6435430C2036
1 changed files with 31 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
crates/lox-distributor/src/main.rs
View File

@ -4,7 +4,6 @@ use hyper::{
body, body,
body::Bytes, body::Bytes,
header::HeaderValue, header::HeaderValue,
http::response,
server::conn::AddrStream, server::conn::AddrStream,
service::{make_service_fn, service_fn}, service::{make_service_fn, service_fn},
Body, Method, Request, Response, Server, StatusCode, Body, Method, Request, Response, Server, StatusCode,
@ -33,7 +32,11 @@ use std::{
use serde_json; use serde_json;
use serde_with::serde_as; 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] #[serde_as]
#[derive(Serialize, Deserialize)] #[derive(Serialize, Deserialize)]
@ -209,7 +212,6 @@ impl LoxServerContext {
} }
} }
async fn handle( async fn handle(
cloned_context: LoxServerContext, cloned_context: LoxServerContext,
req: Request<Body>, req: Request<Body>,
@ -225,7 +227,9 @@ async fn handle(
.unwrap()), .unwrap()),
_ => match (req.method(), req.uri().path()) { _ => match (req.method(), req.uri().path()) {
(&Method::GET, "/invite") => Ok::<_, Infallible>(generate_invite(cloned_context)), (&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::GET, "/pubkeys") => Ok::<_, Infallible>(send_keys(cloned_context)),
(&Method::POST, "/openreq") => Ok::<_, Infallible>({ (&Method::POST, "/openreq") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap(); let bytes = body::to_bytes(req.into_body()).await.unwrap();
@ -386,12 +390,15 @@ async fn parse_bridges(rdsys_tx: mpsc::Sender<Command>, rx: Receiver<ResourceDif
resourcediff: resourcediff, resourcediff: resourcediff,
}; };
rdsys_tx.send(cmd).await.unwrap(); rdsys_tx.send(cmd).await.unwrap();
sleep(Duration::from_secs(1)).await;
} }
} }
#[derive(Debug)] #[derive(Debug)]
enum Command { enum Command {
Rdsys { resourcediff: ResourceDiff }, Rdsys {
resourcediff: ResourceDiff,
},
Request { Request {
req: Request<Body>, req: Request<Body>,
sender: oneshot::Sender<Result<Response<Body>, Infallible>>, sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
@ -408,7 +415,7 @@ async fn main() {
let rtype: ResourceInfo = serde_json::from_reader(reader).unwrap(); let rtype: ResourceInfo = serde_json::from_reader(reader).unwrap();
let (rdsys_tx, mut context_rx) = mpsc::channel(32); 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 { let context_manager = spawn(async move {
// pass in distribution of open invite vs. hot spare buckets? // pass in distribution of open invite vs. hot spare buckets?
@ -512,17 +519,21 @@ async fn main() {
} }
} }
context.encrypt_table(); context.encrypt_table();
sleep(Duration::from_millis(1)).await;
} }
Request { req, sender } => { Request { req, sender } => {
let response = handle(context.clone(), req).await; 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) //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 // to populate the bridge db
let rstream = start_stream(rtype.endpoint, rtype.name, rtype.token, rtype.types) let rstream = start_stream(rtype.endpoint, rtype.name, rtype.token, rtype.types)
.await .await
@ -531,10 +542,9 @@ async fn main() {
let rdsys_resource_receiver = spawn(async { parse_bridges(rdsys_tx, rx).await }); let rdsys_resource_receiver = spawn(async { parse_bridges(rdsys_tx, rx).await });
let make_service = make_service_fn(move |_conn: &AddrStream| {
let make_service = make_service_fn(|_conn: &AddrStream| {
let request_tx = request_tx.clone(); 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 request_tx = request_tx.clone();
let (response_tx, response_rx) = oneshot::channel(); let (response_tx, response_rx) = oneshot::channel();
let cmd = Command::Request { let cmd = Command::Request {
@ -556,9 +566,6 @@ async fn main() {
if let Err(e) = graceful.await { if let Err(e) = graceful.await {
eprintln!("server error: {}", e); eprintln!("server error: {}", e);
} }
future::join_all([rdsys_stream_handler, rdsys_resource_receiver, context_manager]).await;
future::join_all([rdsys_stream_handler, rdsys_resource_receiver]).await;
} }