Refactor API: replace Warp with Axum

Cargo.toml

@@ -58,6 +58,7 @@ async-trait = "0.1"
 
 aquatic_udp_protocol = "0.2"
 uuid = { version = "1", features = ["v4"] }
+axum = "0.6.1"
 
 [dev-dependencies]
 mockall = "0.11"

src/apis/mod.rs

@@ -0,0 +1,2 @@
+pub mod routes;
+pub mod server;

src/apis/routes.rs

@@ -0,0 +1,65 @@
+use std::sync::Arc;
+
+use axum::extract::State;
+use axum::response::Json;
+use serde_json::{json, Value};
+
+use crate::api::resource::stats::Stats;
+use crate::tracker::Tracker;
+
+#[allow(clippy::unused_async)]
+pub async fn root() -> Json<Value> {
+    Json(json!({ "data": 42 }))
+}
+
+#[allow(clippy::unused_async)]
+pub async fn get_stats(State(tracker): State<Arc<Tracker>>) -> Json<Value> {
+    let mut results = Stats {
+        torrents: 0,
+        seeders: 0,
+        completed: 0,
+        leechers: 0,
+        tcp4_connections_handled: 0,
+        tcp4_announces_handled: 0,
+        tcp4_scrapes_handled: 0,
+        tcp6_connections_handled: 0,
+        tcp6_announces_handled: 0,
+        tcp6_scrapes_handled: 0,
+        udp4_connections_handled: 0,
+        udp4_announces_handled: 0,
+        udp4_scrapes_handled: 0,
+        udp6_connections_handled: 0,
+        udp6_announces_handled: 0,
+        udp6_scrapes_handled: 0,
+    };
+
+    let db = tracker.get_torrents().await;
+
+    db.values().for_each(|torrent_entry| {
+        let (seeders, completed, leechers) = torrent_entry.get_stats();
+        results.seeders += seeders;
+        results.completed += completed;
+        results.leechers += leechers;
+        results.torrents += 1;
+    });
+
+    let stats = tracker.get_stats().await;
+
+    #[allow(clippy::cast_possible_truncation)]
+    {
+        results.tcp4_connections_handled = stats.tcp4_connections_handled as u32;
+        results.tcp4_announces_handled = stats.tcp4_announces_handled as u32;
+        results.tcp4_scrapes_handled = stats.tcp4_scrapes_handled as u32;
+        results.tcp6_connections_handled = stats.tcp6_connections_handled as u32;
+        results.tcp6_announces_handled = stats.tcp6_announces_handled as u32;
+        results.tcp6_scrapes_handled = stats.tcp6_scrapes_handled as u32;
+        results.udp4_connections_handled = stats.udp4_connections_handled as u32;
+        results.udp4_announces_handled = stats.udp4_announces_handled as u32;
+        results.udp4_scrapes_handled = stats.udp4_scrapes_handled as u32;
+        results.udp6_connections_handled = stats.udp6_connections_handled as u32;
+        results.udp6_announces_handled = stats.udp6_announces_handled as u32;
+        results.udp6_scrapes_handled = stats.udp6_scrapes_handled as u32;
+    }
+
+    Json(json!(results))
+}

src/apis/server.rs

@@ -0,0 +1,39 @@
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use axum::routing::get;
+use axum::Router;
+use futures::Future;
+use warp::hyper;
+
+use super::routes::{get_stats, root};
+use crate::tracker;
+
+pub fn start(socket_addr: SocketAddr, tracker: &Arc<tracker::Tracker>) -> impl Future<Output = hyper::Result<()>> {
+    let app = Router::new()
+        .route("/", get(root))
+        .route("/stats", get(get_stats).with_state(tracker.clone()));
+
+    let server = axum::Server::bind(&socket_addr).serve(app.into_make_service());
+
+    server.with_graceful_shutdown(async move {
+        tokio::signal::ctrl_c().await.expect("Failed to listen to shutdown signal.");
+    })
+}
+
+pub fn start_tls(
+    socket_addr: SocketAddr,
+    _ssl_cert_path: &str,
+    _ssl_key_path: &str,
+    _tracker: &Arc<tracker::Tracker>,
+) -> impl Future<Output = hyper::Result<()>> {
+    // todo: for the time being, it's just a copy & paste from start(...).
+
+    let app = Router::new().route("/", get(root));
+
+    let server = axum::Server::bind(&socket_addr).serve(app.into_make_service());
+
+    server.with_graceful_shutdown(async move {
+        tokio::signal::ctrl_c().await.expect("Failed to listen to shutdown signal.");
+    })
+}

src/config.rs

@@ -31,7 +31,7 @@ pub struct HttpTracker {
 }
 
 #[serde_as]
-#[derive(Serialize, Deserialize, PartialEq, Eq, Debug)]
+#[derive(Serialize, Deserialize, PartialEq, Eq, Debug, Clone)]
 pub struct HttpApi {
     pub enabled: bool,
     pub bind_address: String,

src/jobs/mod.rs

@@ -1,4 +1,5 @@
 pub mod http_tracker;
 pub mod torrent_cleanup;
 pub mod tracker_api;
+pub mod tracker_apis;
 pub mod udp_tracker;

src/jobs/tracker_apis.rs

@@ -0,0 +1,54 @@
+use std::sync::Arc;
+
+use log::info;
+use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
+
+use crate::apis::server;
+use crate::config::HttpApi;
+use crate::tracker;
+
+#[derive(Debug)]
+pub struct ApiServerJobStarted();
+
+/// # Panics
+///
+/// It would panic if unable to send the  `ApiServerJobStarted` notice.
+pub async fn start_job(config: &HttpApi, tracker: Arc<tracker::Tracker>) -> JoinHandle<()> {
+    let bind_addr = config
+        .bind_address
+        .parse::<std::net::SocketAddr>()
+        .expect("Tracker API bind_address invalid.");
+    let ssl_enabled = config.ssl_enabled;
+    let ssl_cert_path = config.ssl_cert_path.clone();
+    let ssl_key_path = config.ssl_key_path.clone();
+
+    let (tx, rx) = oneshot::channel::<ApiServerJobStarted>();
+
+    // Run the API server
+    let join_handle = tokio::spawn(async move {
+        if !ssl_enabled {
+            info!("Starting Torrust APIs server on: http://{}", bind_addr);
+            let handle = server::start(bind_addr, &tracker);
+            tx.send(ApiServerJobStarted()).expect("the start job dropped");
+            if let Ok(()) = handle.await {
+                info!("Stopping Torrust APIs server on {} ...", bind_addr);
+            }
+        } else if ssl_enabled && ssl_cert_path.is_some() && ssl_key_path.is_some() {
+            info!("Starting Torrust APIs server on: https://{}", bind_addr);
+            let handle = server::start_tls(bind_addr, &ssl_cert_path.unwrap(), &ssl_key_path.unwrap(), &tracker);
+            tx.send(ApiServerJobStarted()).expect("the start job dropped");
+            if let Ok(()) = handle.await {
+                info!("Stopping Torrust APIs server on {} ...", bind_addr);
+            }
+        }
+    });
+
+    // Wait until the APIs server job is running
+    match rx.await {
+        Ok(_msg) => info!("Torrust APIs server started"),
+        Err(e) => panic!("the apis server was dropped: {e}"),
+    }
+
+    join_handle
+}

src/lib.rs

@@ -1,4 +1,5 @@
 pub mod api;
+pub mod apis;
 pub mod config;
 pub mod databases;
 pub mod http;