feat: support for accessing HTTP services internal state

Author: rsfreitas

examples/services/http/Cargo.toml

@@ -8,3 +8,4 @@ axum = "0.7.7"
 mikros = { path = "../../../" }
 tokio = { version = "1.41.1", features = ["full"] }
 tonic = "0.12.3"
+futures = "0.3.31"

examples/services/http/src/main.rs

@@ -1,28 +1,55 @@
+use std::any::Any;
 use std::sync::Arc;
 
 use axum::extract::State;
 use axum::routing::get;
+use mikros::FutureMutex;
 use mikros::http::{ServiceInternalState, ServiceState};
 use mikros::service::builder::ServiceBuilder;
-use mikros::FutureMutex;
+//use mikros::FutureMutex;
 
 #[derive(Clone, Default)]
-pub struct AppState;
-impl ServiceInternalState for AppState {}
+pub struct AppState {
+    value: i32,
+}
+
+impl AppState {
+    pub fn increase(&mut self) {
+        self.value += 1;
+    }
+}
+
+impl ServiceInternalState for AppState {
+    fn clone_box(&self) -> Box<dyn ServiceInternalState> {
+        Box::new(self.clone())
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+}
 
 // Handler method for the first endpoint
 //async fn handler_one(ctx: Option<State<Arc<Context>>>) -> String {
-async fn handler_one(State(state): State<Arc<ServiceState>>) -> String {
+async fn handler_one(State(state): State<Arc<FutureMutex<ServiceState>>>) -> String {
     println!("Handler One");
-    let context = state.context();
+    let context = state.lock().await.context();
     context.logger().info("just a log message");
 
     format!("Handler One")
 }
 
 // Handler method for the second endpoint
-async fn handler_two(State(_state): State<Arc<ServiceState>>) -> String {
+async fn handler_two(State(state): State<Arc<FutureMutex<ServiceState>>>) -> String {
     println!("Handler Two");
+
+    if let Some(app_state) = state.lock().await.state::<AppState>().await {
+        println!("App State current value: {}", app_state.value);
+        app_state.increase();
+        let mut x = app_state.clone_box();
+        x.in
+    }
+
     format!("Handler Two")
 }
 
@@ -34,7 +61,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let state = AppState::default();
     let svc = ServiceBuilder::default()
-        .http_with_state(api, Arc::new(FutureMutex::new(state)))
+        .http_with_state(api, Box::new(state))
+//        .http_with_state(api, Arc::new(FutureMutex::new(state)))
 //        .http(api)
         .build();
 

src/http/mod.rs

@@ -1,37 +1,48 @@
+use std::any::Any;
 use std::sync::Arc;
 
 use futures::lock::Mutex;
 
 use crate::service::context::Context;
 
-pub trait ServiceInternalState: Send + 'static {}
-
 #[derive(Clone)]
 pub struct ServiceState {
     context: Arc<Context>,
-    state: Option<Box<Arc<Mutex<dyn ServiceInternalState>>>>
+    state: Arc<Mutex<Option<Box<dyn ServiceInternalState>>>>
 }
 
 impl ServiceState {
-    pub(crate) fn new(context: &Context) -> ServiceState {
+    pub(crate) fn new(context: &Context) -> Self {
         Self {
             context: Arc::new(context.clone()),
-            state: None
+            state: Arc::new(Mutex::new(None)),
         }
     }
 
-    pub(crate) fn new_with_state(context: &Context, state: &Box<Arc<Mutex<dyn ServiceInternalState>>>) -> ServiceState {
+    pub(crate) fn new_with_state(context: &Context, state: Box<dyn ServiceInternalState>) -> Self {
         Self {
             context: Arc::new(context.clone()),
-            state: Some(state.clone())
+            state: Arc::new(Mutex::new(Some(state))),
         }
     }
 
     pub fn context(&self) -> Arc<Context> {
         self.context.clone()
     }
 
-    pub fn state(&self) -> Option<Box<Arc<Mutex<dyn ServiceInternalState>>>> {
-        self.state.clone()
+    pub async fn state<T: Clone + 'static>(&self) -> Option<Arc<T>> {
+        let state = self.state.lock().await;
+        state.as_ref()?.as_any().downcast_ref::<T>().map(|t|Arc::new(t.clone()))
     }
-}
+}
+
+pub trait ServiceInternalState: Send + Sync + 'static {
+    fn clone_box(&self) -> Box<dyn ServiceInternalState>;
+    fn as_any(&self) -> &dyn Any;
+}
+
+impl Clone for Box<dyn ServiceInternalState> {
+    fn clone(&self) -> Self {
+        self.clone_box()
+    }
+}

src/service/builder.rs

@@ -88,34 +88,30 @@ impl ServiceBuilder {
 
     /// Initializes the HTTP service type with the required structure
     /// implementing the service endpoint handlers.
-    pub fn http(mut self, router: Router<Arc<ServiceState>>) -> Self {
+    pub fn http(mut self, router: Router<Arc<Mutex<ServiceState>>>) -> 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<L>(mut self, router: Router<Arc<ServiceState>>, lifecycle: Arc<Mutex<L>>) -> Self
+    pub fn http_with_lifecycle<L>(mut self, router: Router<Arc<Mutex<ServiceState>>>, lifecycle: Arc<Mutex<L>>) -> Self
     where
         L: Lifecycle + 'static
     {
         self.servers.insert(definition::ServiceKind::Http.to_string(), Box::new(Http::new_with_lifecycle(router, lifecycle)));
         self
     }
 
-    pub fn http_with_state<S>(mut self, router: Router<Arc<ServiceState>>, state: Arc<Mutex<S>>) -> Self
-    where
-        S: ServiceInternalState + 'static,
-    {
+    pub fn http_with_state(mut self, router: Router<Arc<Mutex<ServiceState>>>, state: Box<dyn ServiceInternalState>) -> Self {
         self.servers.insert(definition::ServiceKind::Http.to_string(), Box::new(Http::new_with_state(router, state)));
         self
     }
 
-    pub fn http_with_lifecycle_and_state<L, S>(mut self, router: Router<Arc<ServiceState>>, lifecycle: Arc<Mutex<L>>, state: Arc<Mutex<S>>) -> Self
+    pub fn http_with_lifecycle_and_state<L>(mut self, router: Router<Arc<Mutex<ServiceState>>>, lifecycle: Arc<Mutex<L>>, state: Box<dyn ServiceInternalState>) -> Self
     where
         L: Lifecycle + 'static,
-        S: ServiceInternalState + 'static,
     {
         self.servers.insert(definition::ServiceKind::Http.to_string(), Box::new(Http::new_with_lifecycle_and_state(router, lifecycle, state)));
         self

src/service/http/mod.rs

@@ -1,4 +1,5 @@
 use std::collections::HashMap;
+use std::ops::Deref;
 use std::sync::Arc;
 
 use axum::Router;
@@ -15,22 +16,22 @@ use crate::service::lifecycle::Lifecycle;
 #[derive(Clone)]
 pub(crate) struct Http {
     port: i32,
-    router: Router<Arc<ServiceState>>,
+    router: Router<Arc<Mutex<ServiceState>>>,
     lifecycle: Option<Box<Arc<Mutex<dyn Lifecycle>>>>,
-    internal_state: Option<Box<Arc<Mutex<dyn ServiceInternalState>>>>
+    internal_state: Arc<Mutex<Option<Box<dyn ServiceInternalState>>>>
 }
 
 impl Http {
-    pub fn new(router: Router<Arc<ServiceState>>) -> Self {
+    pub fn new(router: Router<Arc<Mutex<ServiceState>>>) -> Self {
         Self {
             port: 0,
             router,
             lifecycle: None,
-            internal_state: None
+            internal_state: Arc::new(Mutex::new(None))
         }
     }
 
-    pub fn new_with_lifecycle<L>(router: Router<Arc<ServiceState>>, lifecycle: Arc<Mutex<L>>) -> Self
+    pub fn new_with_lifecycle<L>(router: Router<Arc<Mutex<ServiceState>>>, lifecycle: Arc<Mutex<L>>) -> Self
     where
         L: Lifecycle + 'static,
     {
@@ -39,23 +40,19 @@ impl Http {
         s
     }
 
-    pub fn new_with_state<S>(router: Router<Arc<ServiceState>>, state: Arc<Mutex<S>>) -> Self
-    where
-        S: ServiceInternalState + 'static,
-    {
+    pub fn new_with_state(router: Router<Arc<Mutex<ServiceState>>>, state: Box<dyn ServiceInternalState>) -> Self {
         let mut s = Self::new(router);
-        s.internal_state = Some(Box::new(state));
+        s.internal_state = Arc::new(Mutex::new(Some(state)));
         s
     }
 
-    pub fn new_with_lifecycle_and_state<L, S>(router: Router<Arc<ServiceState>>, lifecycle: Arc<Mutex<L>>, state: Arc<Mutex<S>>) -> Self
+    pub fn new_with_lifecycle_and_state<L>(router: Router<Arc<Mutex<ServiceState>>>, lifecycle: Arc<Mutex<L>>, state: Box<dyn ServiceInternalState>) -> Self
     where
         L: Lifecycle + 'static,
-        S: ServiceInternalState + 'static
     {
         let mut s = Self::new(router);
         s.lifecycle = Some(Box::new(lifecycle));
-        s.internal_state = Some(Box::new(state));
+        s.internal_state = Arc::new(Mutex::new(Some(state)));
         s
     }
 }
@@ -111,12 +108,13 @@ impl plugin::service::Service for Http {
             shutdown_rx.changed().await.ok();
         };
 
-        let state = match &self.internal_state {
+        let internal_state = self.internal_state.clone();
+        let state = match internal_state.lock().await.deref() {
             None => ServiceState::new(ctx),
-            Some(st) => ServiceState::new_with_state(ctx, st)
+            Some(state) => ServiceState::new_with_state(ctx, state.clone_box())
         };
 
-        let app = Router::new().merge(self.router.clone()).with_state(Arc::new(state));
+        let app = Router::new().merge(self.router.clone()).with_state(Arc::new(Mutex::new(state)));
 
         match TcpListener::bind(addr).await {
             Ok(incoming) => {