feat: add lifecycle trait calls for services

Author: rsfreitas

Cargo.toml

@@ -8,15 +8,15 @@ async-trait = "0.1.83"
 env-settings = "0.1.11"
 env-settings-derive = "0.1.11"
 env-settings-utils = "0.1.11"
-logger = { git = "https://github.com/rsfreitas/logger" }
+futures = "0.3.31"
 http = "1.1.0"
+logger = { git = "https://github.com/rsfreitas/logger" }
+prost = "0.13.3"
 serde = "1.0.214"
 serde_derive = "1.0.214"
-tonic = { version = "0.12.3", features = ["transport"]}
 tokio = { version = "1.41.1", features = ["full"] }
 toml = "0.8.19"
+tonic = { version = "0.12.3", features = ["transport"]}
 tower = "0.5.1"
 tower-service = "0.3.3"
-validator = { version = "0.19.0", features = ["derive"] }
-futures = "0.3.31"
-prost = "0.13.3"
+validator = { version = "0.19.0", features = ["derive"] }

examples/services/native/Cargo.toml

@@ -4,5 +4,6 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+async-trait = "0.1.83"
 mikros = { path = "../../../" }
 tokio = { version = "1.41.1", features = ["full"] }

examples/services/native/src/main.rs

@@ -2,14 +2,15 @@ mod service;
 
 extern crate mikros;
 
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 
+use mikros::FutureMutex;
 use mikros::service::builder::{ServiceBuilder};
 use service::Service as AppService;
 
 #[tokio::main]
 async fn main() {
-    let s = Arc::new(Mutex::new(AppService::new()));
+    let s = Arc::new(FutureMutex::new(AppService::new()));
     let svc = ServiceBuilder::default()
         .script(s.clone())
         .native(s.clone())

examples/services/native/src/service.rs

@@ -11,6 +11,19 @@ impl Service {
     }
 }
 
+#[async_trait::async_trait]
+impl mikros::service::lifecycle::Lifecycle for Service {
+    async fn on_start(&mut self) -> merrors::Result<()> {
+        println!("lifecycle on_start");
+        Ok(())
+    }
+
+    async fn on_finish(&self) -> merrors::Result<()> {
+        println!("lifecycle on_finish");
+        Ok(())
+    }
+}
+
 impl mikros::service::native::NativeService for Service {
     fn start(&self, ctx: &Context) -> merrors::Result<()> {
         ctx.logger().info("Start native service");

src/lib.rs

@@ -7,3 +7,6 @@ pub mod service;
 
 mod args;
 mod grpc;
+
+// Forward some declarations for services
+pub use futures::lock::Mutex as FutureMutex;

src/plugin/service.rs

@@ -5,9 +5,10 @@ use tokio::sync::watch;
 
 use crate::{definition, env, errors as merrors};
 use crate::service::context::Context;
+use crate::service::lifecycle::Lifecycle;
 
 #[async_trait::async_trait]
-pub trait Service: Send + ServiceClone {
+pub trait Service: ServiceClone + Lifecycle {
     fn kind(&self) -> definition::ServiceKind;
     fn initialize(&mut self, envs: Arc<env::Env>, definitions: Arc<definition::Definitions>) -> merrors::Result<()>;
     fn info(&self) -> HashMap<String, logger::fields::FieldValue>;

src/service/builder.rs

@@ -1,7 +1,8 @@
 use std::collections::HashMap;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc};
 use std::convert::Infallible;
 
+use futures::lock::Mutex;
 use http::{request::Request, response::Response};
 use tonic::body::BoxBody;
 use tonic::server::NamedService;
@@ -10,6 +11,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::lifecycle::Lifecycle;
 use crate::service::Service;
 
 pub struct ServiceBuilder {
@@ -42,8 +44,9 @@ impl ServiceBuilder {
     pub fn grpc<S>(mut self, svc: S) -> Self
     where
         S: tonic::codegen::Service<Request<BoxBody>, Response = Response<BoxBody>, Error = Infallible>
-            + Clone
             + NamedService
+            + Lifecycle
+            + Clone
             + Send
             + Sync
             + 'static,

src/service/grpc/mod.rs

@@ -14,6 +14,7 @@ use crate::{definition, env, plugin};
 use crate::service::context::Context;
 use crate::errors as merrors;
 use crate::grpc;
+use crate::service::lifecycle::Lifecycle;
 
 #[derive(Clone)]
 pub(crate) struct Grpc<S> {
@@ -25,6 +26,7 @@ impl<S> Grpc<S>
 where
     S: tonic::codegen::Service<Request<BoxBody>, Response = Response<BoxBody>, Error = Infallible>
         + NamedService
+        + Lifecycle
         + Clone
         + Send
         + Sync
@@ -39,11 +41,33 @@ where
     }
 }
 
+#[async_trait::async_trait]
+impl<S> Lifecycle for Grpc<S>
+where
+    S: tonic::codegen::Service<Request<BoxBody>, Response = Response<BoxBody>, Error = Infallible>
+        + NamedService
+        + Lifecycle
+        + Clone
+        + Send
+        + Sync
+        + 'static,
+    S::Future: 'static + Send,
+{
+    async fn on_start(&mut self) -> merrors::Result<()> {
+        self.server.on_start().await
+    }
+
+    async fn on_finish(&self) -> merrors::Result<()> {
+        self.server.on_finish().await
+    }
+}
+
 #[async_trait::async_trait]
 impl<S> plugin::service::Service for Grpc<S>
 where
     S: tonic::codegen::Service<Request<BoxBody>, Response = Response<BoxBody>, Error = Infallible>
         + NamedService
+        + Lifecycle
         + Clone
         + Send
         + Sync
@@ -85,7 +109,7 @@ where
         };
 
         let layer = tower::ServiceBuilder::new()
-            .layer(grpc::ContextExtractor::new(&ctx))
+            .layer(grpc::ContextExtractor::new(ctx))
             .into_inner();
 
         if let Err(e) = Server::builder()
@@ -94,7 +118,7 @@ where
             .serve_with_shutdown(addr, shutdown_signal)
             .await
         {
-            return Err(merrors::Error::InternalServiceError(format!("could not initialize grpc server: {}", e.to_string())))
+            return Err(merrors::Error::InternalServiceError(format!("could not initialize grpc server: {}", e)))
         }
 
         Ok(())

src/service/lifecycle.rs

@@ -1,14 +1,12 @@
 use crate::errors as merrors;
 
 #[async_trait::async_trait]
-pub trait Lifecycle {
-    async fn on_start(&self) -> merrors::Result<()> {
-        println!("default on_start");
+pub trait Lifecycle: Send + Sync {
+    async fn on_start(&mut self) -> merrors::Result<()> {
         Ok(())
     }
 
     async fn on_finish(&self) -> merrors::Result<()> {
-        println!("default on_finish");
         Ok(())
     }
 }

src/service/mod.rs

@@ -3,8 +3,7 @@ pub mod context;
 pub mod grpc;
 pub mod native;
 pub mod script;
-
-mod lifecycle;
+pub mod lifecycle;
 
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -81,7 +80,7 @@ impl Service {
         self.logger.info("service starting");
         self.validate_definitions()?;
         self.start_features()?;
-        self.initialize_service_internals()?;
+        self.initialize_service_internals().await?;
         self.print_service_resources();
         self.run().await
     }
@@ -97,7 +96,7 @@ impl Service {
         }
 
         for t in &self.definitions.types {
-            if self.servers.get(&t.0.to_string()).is_none() {
+            if !self.servers.contains_key(&t.0.to_string()) {
                 return Err(merrors::Error::ServiceKindUninitialized(t.0.clone()))
             }
         }
@@ -110,7 +109,7 @@ impl Service {
         Ok(())
     }
 
-    fn initialize_service_internals(&mut self) -> merrors::Result<()> {
+    async fn initialize_service_internals(&mut self) -> merrors::Result<()> {
         let definitions = self.definitions.clone();
         let envs = self.envs.clone();
 
@@ -119,8 +118,12 @@ impl Service {
             svc.initialize(envs.clone(), definitions.clone())?
         }
 
-        // couple clients
-        // call lifecycle on_start
+        // TODO couple clients
+
+        for s in &definitions.types {
+            let svc = self.get_server(&s.0)?;
+            svc.on_start().await?;
+        }
 
         Ok(())
     }
@@ -200,6 +203,7 @@ impl Service {
         for s in &definitions.types {
             let svc = self.get_server(&s.0)?;
             svc.stop(&context).await;
+            svc.on_finish().await?;
         }
 
         // Then stops our task runner

src/service/native/mod.rs

@@ -1,14 +1,16 @@
 use std::collections::HashMap;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc};
 
+use futures::lock::Mutex;
 use logger::fields::FieldValue;
 use tokio::sync::watch;
 
 use crate::{definition, env, errors as merrors, plugin};
 use crate::service::context::Context;
+use crate::service::lifecycle::Lifecycle;
 
 #[async_trait::async_trait]
-pub trait NativeService: Send + NativeServiceClone {
+pub trait NativeService: NativeServiceClone + Lifecycle + Send + Sync {
     /// This is the place where the service/application must be initialized. It
     /// should do the required initialization, put any job to execute in background
     /// and leave. It shouldn't block.
@@ -52,6 +54,17 @@ impl Native {
     }
 }
 
+#[async_trait::async_trait]
+impl Lifecycle for Native {
+    async fn on_start(&mut self) -> merrors::Result<()> {
+        self.svc.lock().await.on_start().await
+    }
+
+    async fn on_finish(&self) -> merrors::Result<()> {
+        self.svc.lock().await.on_finish().await
+    }
+}
+
 #[async_trait::async_trait]
 impl plugin::service::Service for Native {
     fn kind(&self) -> definition::ServiceKind {
@@ -69,10 +82,10 @@ impl plugin::service::Service for Native {
     }
 
     async fn run(&self, ctx: &Context, _: watch::Receiver<()>) -> merrors::Result<()> {
-        self.svc.lock().unwrap().start(ctx)
+        self.svc.lock().await.start(ctx)
     }
 
     async fn stop(&self, ctx: &Context) {
-        self.svc.lock().unwrap().stop(ctx)
+        self.svc.lock().await.stop(ctx)
     }
 }

src/service/script/mod.rs

@@ -1,13 +1,16 @@
 use std::collections::HashMap;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc};
 
+use futures::lock::Mutex;
 use logger::fields::FieldValue;
 use tokio::sync::watch;
 
 use crate::{definition, env, errors as merrors, plugin};
 use crate::service::context::Context;
+use crate::service::lifecycle::Lifecycle;
 
-pub trait ScriptService: Send + ScriptServiceClone {
+#[async_trait::async_trait]
+pub trait ScriptService: ScriptServiceClone + Lifecycle + Send {
     fn run(&self, ctx: &Context) -> merrors::Result<()>;
     fn cleanup(&self, ctx: &Context);
 }
@@ -44,6 +47,17 @@ impl Script {
     }
 }
 
+#[async_trait::async_trait]
+impl Lifecycle for Script {
+    async fn on_start(&mut self) -> merrors::Result<()> {
+        self.svc.lock().await.on_start().await
+    }
+
+    async fn on_finish(&self) -> merrors::Result<()> {
+        self.svc.lock().await.on_finish().await
+    }
+}
+
 #[async_trait::async_trait]
 impl plugin::service::Service for Script {
     fn kind(&self) -> definition::ServiceKind {
@@ -61,10 +75,10 @@ impl plugin::service::Service for Script {
     }
 
     async fn run(&self, ctx: &Context, _: watch::Receiver<()>) -> merrors::Result<()> {
-        self.svc.lock().unwrap().run(ctx)
+        self.svc.lock().await.run(ctx)
     }
 
     async fn stop(&self, ctx: &Context) {
-        self.svc.lock().unwrap().cleanup(ctx)
+        self.svc.lock().await.cleanup(ctx)
     }
-}
+}