feat: initial support for HTTP services

Author: rsfreitas

Cargo.toml

@@ -5,6 +5,7 @@ edition = "2021"
 
 [dependencies]
 async-trait = "0.1.83"
+axum = "0.7.7"
 env-settings = "0.1.11"
 env-settings-derive = "0.1.11"
 env-settings-utils = "0.1.11"

examples/services/http/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "http"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+axum = "0.7.7"
+mikros = { path = "../../../" }
+tokio = { version = "1.41.1", features = ["full"] }
+tonic = "0.12.3"

examples/services/http/service.toml

@@ -0,0 +1,5 @@
+name = "http-example"
+version = "v0.1.0"
+language = "rust"
+product = "examples"
+types = ["http"]

examples/services/http/src/main.rs

@@ -0,0 +1,52 @@
+use std::sync::Arc;
+
+use axum::extract::State;
+use axum::routing::get;
+use mikros::service::builder::ServiceBuilder;
+use mikros::service::context::Context;
+
+#[derive(Clone)]
+pub struct Service {
+}
+
+impl Service {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+// Handler method for the first endpoint
+//async fn handler_one(ctx: Option<State<Arc<Context>>>) -> String {
+async fn handler_one() -> String {
+    println!("Handler One");
+    format!("Handler One")
+}
+
+// Handler method for the second endpoint
+async fn handler_two(State(ctx): State<Arc<Context>>) -> String {
+    println!("Handler Two");
+    format!("Handler Two")
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let x = axum::Router::new()
+        .route("/one", get(handler_one));
+//        .route("/two", get(handler_two));
+    
+    let service = Service::new();
+    let svc = ServiceBuilder::default()
+        .http(x)
+        .build();
+
+    match svc {
+        Ok(mut svc) => {
+            if let Err(e) = svc.start().await {
+                println!("application error: {}", e);
+            }
+        },
+        Err(e) => panic!("{}", e.to_string())
+    }
+
+    Ok(())
+}

src/definition/mod.rs

@@ -121,7 +121,7 @@ impl Definitions {
     }
 
     pub(crate) fn get_service_type(&self, kind: ServiceKind) -> merrors::Result<&service::Service> {
-        match self.types.iter().find(|t| t.0 == definition::ServiceKind::Grpc) {
+        match self.types.iter().find(|t| t.0 == kind) {
             Some(t) => Ok(t),
             None => Err(merrors::Error::NotFound(format!("could not find service kind '{}'", kind)))
         }

src/service/builder.rs

@@ -2,6 +2,7 @@ use std::collections::HashMap;
 use std::sync::{Arc};
 use std::convert::Infallible;
 
+use axum::Router;
 use futures::lock::Mutex;
 use http::{request::Request, response::Response};
 use tonic::body::BoxBody;
@@ -11,6 +12,7 @@ use crate::{definition, errors as merrors, plugin};
 use crate::service::native::{NativeService, Native};
 use crate::service::script::{ScriptService, Script};
 use crate::service::grpc::Grpc;
+use crate::service::http::Http;
 use crate::service::lifecycle::Lifecycle;
 use crate::service::Service;
 
@@ -43,7 +45,7 @@ impl ServiceBuilder {
         self
     }
 
-    /// Initializes the grpc service type with the required structure
+    /// Initializes the gRPC service type with the required structure
     /// implementing its API.
     pub fn grpc<S>(mut self, server: S) -> Self
     where
@@ -59,7 +61,7 @@ impl ServiceBuilder {
         self
     }
 
-    /// Initializes the grpc service type with the required structure
+    /// Initializes the gRPC service type with the required structure
     /// implementing its API and another with implementing the Lifecycle
     /// API.
     pub fn grpc_with_lifecycle<S, B: Lifecycle + 'static>(mut self, server: S, lifecycle: Arc<Mutex<B>>) -> Self
@@ -76,6 +78,21 @@ impl ServiceBuilder {
         self
     }
 
+    /// Initializes the HTTP service type with the required structure
+    /// implementing the service endpoint handlers.
+    pub fn http(mut self, router: Router) -> Self {
+        self.servers.insert(definition::ServiceKind::Http.to_string(), Box::new(Http::new(router)));
+        self
+    }
+
+    /// Initializes the HTTP service type with the required structure
+    /// implementing the service endpoint handlers and another with
+    /// implementing the Lifecycle API.
+    pub fn http_with_lifecycle<B: Lifecycle + 'static>(mut self, router: Router, lifecycle: Arc<Mutex<B>>) -> Self {
+        self.servers.insert(definition::ServiceKind::Http.to_string(), Box::new(Http::new_with_lifecycle(router, lifecycle)));
+        self
+    }
+
     /// Adds external features into the current service environment.
     pub fn with_features(mut self, features: Vec<Box<dyn plugin::feature::Feature>>) -> Self {
         self.features.extend(features);

src/service/http/mod.rs

@@ -0,0 +1,108 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use axum::Router;
+use futures::lock::Mutex;
+use logger::fields::FieldValue;
+use tokio::net::TcpListener;
+use tokio::sync::watch::Receiver;
+
+use crate::{definition, env, errors as merrors, plugin};
+use crate::service::context::Context;
+use crate::service::lifecycle::Lifecycle;
+
+#[derive(Clone)]
+pub(crate) struct Http {
+    port: i32,
+    router: Router,
+    lifecycle: Option<Box<Arc<Mutex<dyn Lifecycle>>>>,
+}
+
+impl Http {
+    pub fn new(router: Router) -> Self {
+        Self {
+            port: 0,
+            router,
+            lifecycle: None,
+        }
+    }
+
+    pub fn new_with_lifecycle<B: Lifecycle + 'static>(router: Router, lifecycle: Arc<Mutex<B>>) -> Self {
+        Self {
+            port: 0,
+            router,
+            lifecycle: Some(Box::new(lifecycle)),
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl Lifecycle for Http {
+    async fn on_start(&mut self) -> merrors::Result<()> {
+        if let Some(lifecycle) = &self.lifecycle {
+            return lifecycle.lock().await.on_start().await
+        }
+
+        Ok(())
+    }
+
+    async fn on_finish(&self) -> merrors::Result<()> {
+        if let Some(lifecycle) = &self.lifecycle {
+            return lifecycle.lock().await.on_finish().await
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl plugin::service::Service for Http {
+    fn kind(&self) -> definition::ServiceKind {
+        definition::ServiceKind::Http
+    }
+
+    fn initialize(&mut self, envs: Arc<env::Env>, definitions: Arc<definition::Definitions>) -> merrors::Result<()> {
+        let service_type = definitions.get_service_type(definition::ServiceKind::Http)?;
+        self.port = match service_type.1 {
+            None => envs.http_port,
+            Some(port) => port,
+        };
+
+        Ok(())
+    }
+
+    fn info(&self) -> HashMap<String, FieldValue> {
+        logger::fields![
+            "svc.port" => FieldValue::Number(self.port as i64),
+            "svc.mode" => FieldValue::String(definition::ServiceKind::Http.to_string()),
+        ]
+    }
+
+    async fn run(&self, ctx: &Context, shutdown_rx: Receiver<()>) -> merrors::Result<()> {
+        let addr = format!("0.0.0.0:{}", self.port);
+        let shutdown_signal = async move {
+            let mut shutdown_rx = shutdown_rx.clone();
+
+            // Wait until the receiver sees the shutdown signal
+            shutdown_rx.changed().await.ok();
+        };
+
+        let shared_ctx = Arc::new(ctx.clone());
+        let app = Router::new().with_state(shared_ctx).merge(self.router.clone());
+
+        match TcpListener::bind(addr).await {
+            Ok(incoming) => {
+                if let Err(e) = axum::serve(incoming, app).with_graceful_shutdown(shutdown_signal).await {
+                    return Err(merrors::Error::InternalServiceError(format!("could not initialize http server: {}", e)))
+                }
+
+                Ok(())
+            }
+            Err(e) => Err(merrors::Error::InternalServiceError(format!("could not initialize http server: {}", e)))
+        }
+    }
+
+    async fn stop(&self, _ctx: &Context) {
+        // noop
+    }
+}

src/service/mod.rs

@@ -1,9 +1,10 @@
 pub mod builder;
 pub mod context;
 pub mod grpc;
+pub mod http;
+pub mod lifecycle;
 pub mod native;
 pub mod script;
-pub mod lifecycle;
 
 use std::collections::HashMap;
 use std::sync::Arc;