2024-03-23 17:43:18 -04:00
|
|
|
use troll_patrol::{request_handler::handle, *};
|
2024-02-07 18:42:08 -05:00
|
|
|
|
|
|
|
|
use clap::Parser;
|
2024-03-22 23:42:22 -04:00
|
|
|
use futures::future;
|
|
|
|
|
use hyper::{
|
|
|
|
|
server::conn::AddrStream,
|
|
|
|
|
service::{make_service_fn, service_fn},
|
|
|
|
|
Body, Request, Response, Server,
|
|
|
|
|
};
|
|
|
|
|
use serde::Deserialize;
|
2024-02-07 18:42:08 -05:00
|
|
|
use sled::Db;
|
2024-03-22 23:42:22 -04:00
|
|
|
use std::{
|
2024-03-23 17:43:18 -04:00
|
|
|
collections::BTreeMap, convert::Infallible, fs::File, io::BufReader, net::SocketAddr,
|
2024-03-22 23:42:22 -04:00
|
|
|
path::PathBuf, time::Duration,
|
|
|
|
|
};
|
|
|
|
|
use tokio::{
|
|
|
|
|
signal, spawn,
|
|
|
|
|
sync::{broadcast, mpsc, oneshot},
|
|
|
|
|
time::sleep,
|
|
|
|
|
};
|
2024-02-07 18:42:08 -05:00
|
|
|
|
2024-03-22 23:42:22 -04:00
|
|
|
async fn shutdown_signal() {
|
|
|
|
|
tokio::signal::ctrl_c()
|
|
|
|
|
.await
|
|
|
|
|
.expect("failed to listen for ctrl+c signal");
|
|
|
|
|
println!("Shut down Troll Patrol Server");
|
|
|
|
|
}
|
2024-02-07 18:42:08 -05:00
|
|
|
|
2024-03-22 23:42:22 -04:00
|
|
|
#[derive(Parser, Debug)]
|
|
|
|
|
#[command(author, version, about, long_about = None)]
|
|
|
|
|
struct Args {
|
|
|
|
|
/// Name/path of the configuration file
|
|
|
|
|
#[arg(short, long, default_value = "config.json")]
|
|
|
|
|
config: PathBuf,
|
|
|
|
|
}
|
2024-02-07 18:42:08 -05:00
|
|
|
|
2024-03-22 23:42:22 -04:00
|
|
|
#[derive(Debug, Deserialize)]
|
|
|
|
|
pub struct Config {
|
|
|
|
|
pub db: DbConfig,
|
2024-03-23 17:43:18 -04:00
|
|
|
// map of distributor name to IP:port to contact it
|
|
|
|
|
pub distributors: BTreeMap<BridgeDistributor, String>,
|
2024-03-22 23:42:22 -04:00
|
|
|
//require_bridge_token: bool,
|
|
|
|
|
port: u16,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
|
|
|
pub struct DbConfig {
|
|
|
|
|
// The path for the server database, default is "server_db"
|
|
|
|
|
pub db_path: String,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
impl Default for DbConfig {
|
|
|
|
|
fn default() -> DbConfig {
|
|
|
|
|
DbConfig {
|
|
|
|
|
db_path: "server_db".to_owned(),
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2024-03-23 17:43:18 -04:00
|
|
|
async fn update_daily_info(db: &Db, distributors: &BTreeMap<BridgeDistributor, String>) {
|
|
|
|
|
update_extra_infos(&db).await;
|
|
|
|
|
update_negative_reports(&db, &distributors).await;
|
|
|
|
|
update_positive_reports(&db, &distributors).await;
|
2024-03-23 23:16:20 -04:00
|
|
|
let new_blockages = guess_blockages(&db, &analyzer::ExampleAnalyzer {});
|
|
|
|
|
report_blockages(&distributors, new_blockages).await;
|
2024-03-22 23:42:22 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async fn create_context_manager(
|
|
|
|
|
db_config: DbConfig,
|
|
|
|
|
context_rx: mpsc::Receiver<Command>,
|
|
|
|
|
mut kill: broadcast::Receiver<()>,
|
|
|
|
|
) {
|
|
|
|
|
tokio::select! {
|
|
|
|
|
create_context = context_manager(db_config, context_rx) => create_context,
|
|
|
|
|
_ = kill.recv() => {println!("Shut down manager");},
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async fn context_manager(db_config: DbConfig, mut context_rx: mpsc::Receiver<Command>) {
|
|
|
|
|
let db: Db = sled::open(&db_config.db_path).unwrap();
|
|
|
|
|
|
|
|
|
|
while let Some(cmd) = context_rx.recv().await {
|
|
|
|
|
use Command::*;
|
|
|
|
|
match cmd {
|
|
|
|
|
Request { req, sender } => {
|
|
|
|
|
let response = handle(&db, req).await;
|
|
|
|
|
if let Err(e) = sender.send(response) {
|
|
|
|
|
eprintln!("Server Response Error: {:?}", e);
|
|
|
|
|
};
|
|
|
|
|
sleep(Duration::from_millis(1)).await;
|
|
|
|
|
}
|
|
|
|
|
Shutdown { shutdown_sig } => {
|
|
|
|
|
println!("Sending Shutdown Signal, all threads should shutdown.");
|
|
|
|
|
drop(shutdown_sig);
|
|
|
|
|
println!("Shutdown Sent.");
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Each of the commands that can be handled
|
|
|
|
|
#[derive(Debug)]
|
|
|
|
|
enum Command {
|
|
|
|
|
Request {
|
|
|
|
|
req: Request<Body>,
|
|
|
|
|
sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
|
|
|
|
|
},
|
|
|
|
|
Shutdown {
|
|
|
|
|
shutdown_sig: broadcast::Sender<()>,
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[tokio::main]
|
|
|
|
|
async fn main() {
|
|
|
|
|
// TODO: Currently, we're processing extra-infos here, but we want to:
|
|
|
|
|
// 2. Periodically (daily):
|
|
|
|
|
// a) download new extra-infos
|
|
|
|
|
// b) determine whether we think each bridge is blocked or not
|
|
|
|
|
// c) report these results to the LA
|
|
|
|
|
// 3. Store all our data
|
|
|
|
|
|
|
|
|
|
let args: Args = Args::parse();
|
|
|
|
|
|
|
|
|
|
let config: Config = serde_json::from_reader(BufReader::new(
|
|
|
|
|
File::open(&args.config).expect("Could not read config file"),
|
|
|
|
|
))
|
|
|
|
|
.expect("Reading config file from JSON failed");
|
|
|
|
|
|
|
|
|
|
let (request_tx, request_rx) = mpsc::channel(32);
|
|
|
|
|
|
|
|
|
|
let shutdown_cmd_tx = request_tx.clone();
|
|
|
|
|
|
|
|
|
|
// create the shutdown broadcast channel and clone for every thread
|
|
|
|
|
let (shutdown_tx, mut shutdown_rx) = broadcast::channel(16);
|
|
|
|
|
let kill = shutdown_tx.subscribe();
|
|
|
|
|
|
|
|
|
|
// 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() => {
|
|
|
|
|
let cmd = Command::Shutdown {
|
|
|
|
|
shutdown_sig: shutdown_tx,
|
|
|
|
|
};
|
|
|
|
|
shutdown_cmd_tx.send(cmd).await.unwrap();
|
|
|
|
|
sleep(Duration::from_secs(1)).await;
|
|
|
|
|
|
|
|
|
|
_ = shutdown_rx.recv().await;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
let context_manager =
|
|
|
|
|
spawn(async move { create_context_manager(config.db, request_rx, kill).await });
|
|
|
|
|
|
|
|
|
|
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();
|
|
|
|
|
let (response_tx, response_rx) = oneshot::channel();
|
|
|
|
|
let cmd = Command::Request {
|
|
|
|
|
req,
|
|
|
|
|
sender: response_tx,
|
|
|
|
|
};
|
|
|
|
|
async move {
|
|
|
|
|
request_tx.send(cmd).await.unwrap();
|
|
|
|
|
response_rx.await.unwrap()
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
async move { Ok::<_, Infallible>(service) }
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
let addr = SocketAddr::from(([0, 0, 0, 0], config.port));
|
|
|
|
|
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([context_manager, shutdown_handler]).await;
|
2024-02-07 18:42:08 -05:00
|
|
|
}
|
