feat: program on layer

Cargo.toml

@@ -25,5 +25,6 @@ serde = { version = "1.0.215", features = ["derive"] }
 axum = { version = "0.7" }
 thiserror = { version = "1.0" }
 tower-http = { version = "0.6.2", features = ["trace"] }
+tower = { version = "0.5.2", features = ["util"] }
 lumx_core = { path = "./lumx_core" }
 lumx_axum = { path = "./lumx_axum" }

lumx_axum/Cargo.toml

@@ -14,3 +14,4 @@ tokio = { workspace = true }
 tracing = { workspace = true }
 serde = { workspace = true }
 tower-http = { workspace = true, features = ["trace"] }
+tower = { workspace = true }

lumx_axum/src/lib.rs

@@ -1,6 +1,7 @@
 pub mod router;
 pub use axum;
 pub mod extractor;
+pub mod middleware;
 pub mod plugin;
 pub mod program;
 pub mod state;

lumx_axum/src/middleware/mod.rs

@@ -0,0 +1 @@
+pub mod state;

lumx_axum/src/middleware/state.rs

@@ -0,0 +1,52 @@
+use axum::extract::Request;
+use lumx_core::program::Program;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use tower::{Layer, Service};
+
+#[derive(Clone)]
+pub struct StateLayer {
+    program: Arc<Program>,
+}
+
+impl StateLayer {
+    pub fn new(program: Arc<Program>) -> Self {
+        Self { program }
+    }
+}
+
+impl<S> Layer<S> for StateLayer {
+    type Service = StateLayerService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        StateLayerService {
+            inner,
+            program: self.program.clone(),
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct StateLayerService<S> {
+    inner: S,
+    program: Arc<Program>,
+}
+
+impl<S, B> Service<Request<B>> for StateLayerService<S>
+where
+    S: Service<Request<B>>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, mut req: Request<B>) -> Self::Future {
+        req.extensions_mut().insert(self.program.clone());
+
+        self.inner.call(req)
+    }
+}

lumx_axum/src/plugin.rs

@@ -1,4 +1,3 @@
-use crate::{router::RouterRef, state::AppState};
 use axum::{async_trait, Extension, Router};
 use lumx_core::{
     plugable::plugin::Plugin,
@@ -10,6 +9,9 @@ use std::{net::SocketAddr, sync::Arc};
 use tower_http::trace::TraceLayer;
 use tracing::debug;
 
+use crate::middleware::state::StateLayer;
+use crate::{router::RouterRef, state::AppState};
+
 pub struct WebPlugin;
 
 #[async_trait]
@@ -35,7 +37,10 @@ impl WebPlugin {
 
         debug!(?router, "registered routes");
         let router = router
-            .layer(Extension(AppState { app }))
+            .layer(Extension(AppState {
+                app: Arc::clone(&app),
+            }))
+            .layer(StateLayer::new(Arc::clone(&app)))
             .layer(TraceLayer::new_for_http());
 
         println!("Listening on {}", listener.local_addr().unwrap());

lumx_core/src/lib.rs

@@ -1,7 +1,7 @@
 pub mod plugable;
 pub mod program;
 pub mod scheduler;
-pub mod tracing;
+pub mod tracer;
 pub mod types;
 
 pub use async_trait;

lumx_core/src/program.rs

@@ -9,6 +9,7 @@ use crate::{
         plugin::{Plugin, PluginRef},
     },
     scheduler::Scheduler,
+    tracer,
     types::ProgramFailure,
 };
 
@@ -217,10 +218,13 @@ impl ProgramBuilder {
         // 1. read env variables
         dotenvy::dotenv().ok();
 
-        // 2. build plugins
+        // 2. init and subscribe tracer
+        tracer::init();
+
+        // 3. build plugins
         self.build_plugins().await;
 
-        // 3. schedule
+        // 4. schedule
         self.schedule().await
     }
 
@@ -230,10 +234,13 @@ impl ProgramBuilder {
         // 1. read env variables
         dotenvy::dotenv().ok();
 
-        // 2. build plugins
+        // 2. init and subscribe tracer
+        tracer::init();
+
+        // 3. build plugins
         self.build_plugins().await;
 
-        // 3. build program
+        // 4. build program
         self.build_program()
     }
 

lumx_core/src/tracer.rs

@@ -0,0 +1,11 @@
+use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
+
+pub fn init() {
+    tracing_subscriber::registry()
+        .with(
+            EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| format!("{}=info", env!("CARGO_CRATE_NAME")).into()),
+        )
+        .with(fmt::layer())
+        .init();
+}