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 3b0c8ab5b6
commit c74da3a078
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

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

@ -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<Body>,
@ -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,12 +390,15 @@ async fn parse_bridges(rdsys_tx: mpsc::Sender<Command>, rx: Receiver<ResourceDif
resourcediff: resourcediff,
};
rdsys_tx.send(cmd).await.unwrap();
sleep(Duration::from_secs(1)).await;
}
}
#[derive(Debug)]
enum Command {
Rdsys { resourcediff: ResourceDiff },
Rdsys {
resourcediff: ResourceDiff,
},
Request {
req: Request<Body>,
sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
@ -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,8 +542,7 @@ 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 request_tx = request_tx.clone();
@ -556,9 +566,6 @@ async fn main() {
if let Err(e) = graceful.await {
eprintln!("server error: {}", e);
}
future::join_all([rdsys_stream_handler, rdsys_resource_receiver]).await;
future::join_all([rdsys_stream_handler, rdsys_resource_receiver, context_manager]).await;
}