feat: HTTP API framework

Cargo.toml

@@ -14,9 +14,11 @@ rust-version = "1.79"
 
 [workspace.dependencies]
 arch-api = { path = "crates/api" }
-arch-core = { path = "crates/core"}
+arch-core = { path = "crates/core" }
 
 async-trait = "0.1.81"
+thiserror = "1.0.63"
+tokio-graceful-shutdown = "0.15.0"
 
 [workspace.dependencies.tokio]
 version = "1.39.1"

crates/api/Cargo.toml

@@ -10,3 +10,27 @@ readme.workspace = true
 rust-version.workspace = true
 
 [dependencies]
+arch-core.workspace = true
+
+thiserror.workspace = true
+tokio.workspace = true
+tokio-graceful-shutdown.workspace = true
+tracing.workspace = true
+
+axum = "0.7.5"
+
+[dependencies.hyper]
+version = "1.3.1"
+features = ["full"]
+
+[dependencies.hyper-util]
+version = "0.1.3"
+features = ["tokio", "server-auto", "http1"]
+
+[dependencies.tower]
+version = "0.4.13"
+features = ["util"]
+
+[dependencies.tower-http]
+version = "0.5.2"
+features = ["timeout", "trace"]

crates/api/src/api_error.rs

@@ -0,0 +1,5 @@
+#[derive(Debug, thiserror::Error)]
+pub enum ApiError {
+    #[error("Bad port - {}", _0)]
+    BadPort(u16),
+}

crates/api/src/lib.rs

@@ -1,14 +1,91 @@
-pub fn add(left: u64, right: u64) -> u64 {
-    left + right
+use std::{net::SocketAddr, sync::Arc, time::Duration};
+
+use arch_core::ArchService;
+
+pub mod api_error;
+
+pub use api_error::*;
+use axum::{
+    extract::{Request, State},
+    response::{IntoResponse, Response},
+    routing::get,
+    Router,
+};
+use hyper::{body::Incoming, StatusCode};
+use hyper_util::rt::TokioIo;
+use tokio::net::{TcpListener, TcpStream};
+use tokio_graceful_shutdown::{SubsystemBuilder, SubsystemHandle};
+use tower::Service;
+use tower_http::timeout::TimeoutLayer;
+
+pub async fn start_http(port: u16, arch_service: Arc<ArchService>, subsys: SubsystemHandle) -> Result<(), ApiError> {
+    tracing::trace!("Starting http service");
+
+    let addr: SocketAddr = format!("0.0.0.0:{}", port).parse().map_err(|_| ApiError::BadPort(port))?;
+    let listener = TcpListener::bind(addr).await.unwrap();
+
+    let routes = get_routes(arch_service.clone());
+
+    tracing::info!("Listening on port {}", port);
+    loop {
+        let (socket, remote_addr) = tokio::select! {
+            _ = subsys.on_shutdown_requested() => {
+                break;
+            }
+
+            result = listener.accept() => {
+                result.unwrap()
+            }
+        };
+
+        tracing::debug!("connection {} accepted", remote_addr);
+        let tower_service = routes.clone();
+        let name = format!("handler-{remote_addr}");
+        subsys.start(SubsystemBuilder::new(name, move |h| handler(socket, remote_addr, tower_service, h)));
+    }
+
+    Ok(())
 }
 
-#[cfg(test)]
-mod tests {
-    use super::*;
+fn get_routes(arch_service: Arc<ArchService>) -> Router<()> {
+    axum::Router::new()
+        .route("/health", get(health))
+        .with_state(arch_service)
+        .layer(TimeoutLayer::new(Duration::from_secs(2)))
+}
+
+pub async fn handler(socket: TcpStream, remote_addr: SocketAddr, tower_service: Router<()>, subsys: SubsystemHandle) -> Result<(), ApiError> {
+    let socket = TokioIo::new(socket);
+    let hyper_service = hyper::service::service_fn(move |request: Request<Incoming>| tower_service.clone().call(request));
+    let conn = hyper::server::conn::http1::Builder::new().serve_connection(socket, hyper_service);
+    let mut conn = std::pin::pin!(conn);
+
+    tokio::select! {
+        result = conn.as_mut() => {
+            if let Err(err) = result {
+                tracing::warn!("Failed to serve connection{}: {:#}", remote_addr, err);
+            }
+        }
+
+        _ = subsys.on_shutdown_requested() => {
+            tracing::debug!("signal received, starting graceful shutdown");
+        }
+    }
+
+    tracing::debug!("Connection {} closed", remote_addr);
+    Ok(())
+}
+
+#[tracing::instrument(level = "trace", skip(arch_service))]
+pub async fn health(State(arch_service): State<Arc<ArchService>>) -> impl IntoResponse {
+    match arch_service.health_service.is_healthy().await {
+        true => StatusCode::OK,
+        false => StatusCode::BAD_REQUEST,
+    }
+}
 
-    #[test]
-    fn it_works() {
-        let result = add(2, 2);
-        assert_eq!(result, 4);
+impl IntoResponse for ApiError {
+    fn into_response(self) -> Response {
+        (StatusCode::INTERNAL_SERVER_ERROR, format!("Something went wrong: {}", self)).into_response()
     }
 }

crates/apps/Cargo.toml

@@ -15,9 +15,11 @@ anyhow = "1.0.86"
 log = "0.4.22"
 tracing-log = "0.2.0"
 
+arch-api.workspace = true
 arch-core.workspace = true
 
 tokio.workspace = true
+tokio-graceful-shutdown.workspace = true
 tracing.workspace = true
 
 [dependencies.tracing-subscriber]

crates/apps/src/main.rs

@@ -1,16 +1,28 @@
+use std::{sync::Arc, time::Duration};
+
+use arch_api::start_http;
 use arch_core::create_service;
 
 use anyhow::Result;
+use tokio_graceful_shutdown::{SubsystemBuilder, Toplevel};
 
 mod logging;
 
 #[tokio::main]
 async fn main() -> Result<()> {
     logging::init_logging()?;
 
-    let arch_service = create_service();
+    let arch_service = Arc::new(create_service());
     tracing::info!("Hello, world!");
     tracing::info!("Healthy={}", arch_service.health_service.is_healthy().await);
 
+    let server = Toplevel::new(|s| async move {
+        s.start(SubsystemBuilder::new("http_api", |h| start_http(3000, arch_service, h)));
+    })
+    .catch_signals()
+    .handle_shutdown_requests(Duration::from_secs(5));
+
+    server.await?;
+
     Ok(())
 }

crates/core/Cargo.toml

@@ -11,3 +11,4 @@ rust-version.workspace = true
 
 [dependencies]
 async-trait.workspace = true
+tracing.workspace = true

crates/core/src/health.rs

@@ -2,10 +2,12 @@ use async_trait::async_trait;
 
 use crate::HealthService;
 
+#[derive(Clone)]
 pub(crate) struct HealthServiceImpl {}
 
 #[async_trait]
 impl HealthService for HealthServiceImpl {
+    #[tracing::instrument(level = "trace", skip(self))]
     async fn is_healthy(&self) -> bool {
         true
     }

crates/core/src/lib.rs

@@ -4,7 +4,7 @@ use health::HealthServiceImpl;
 mod health;
 
 #[async_trait]
-pub trait HealthService {
+pub trait HealthService: Send + Sync {
     async fn is_healthy(&self) -> bool;
 }