Merge pull request #390 from code0-tech/#379-send-all-available-modules

send all available modules

Author: raphael-goetz

Cargo.lock

@@ -11,8 +11,8 @@ opentelemetry = { version = "0.32.0", features = ["metrics"] }
 tracing = { version = "0.1.41", features = ["log"] }
 prost = "0.14.1"
 tonic = "0.14.1"
-tucana = { version = "0.0.75", features = ["all"] }
-code0-flow = { version = "0.0.40", features = ["flow_config", "flow_health", "flow_telemetry"] }
+tucana = { version = "0.0.76", features = ["all"] }
+code0-flow = { version = "0.0.41", features = ["flow_config", "flow_health", "flow_telemetry"] }
 serde_json = "1.0.140"
 async-nats = "0.49.0"
 tonic-health = "0.14.1"

Cargo.toml

@@ -3,7 +3,8 @@
 	"runtimes": [
 		{
 			"identifier": "taurus",
-			"token": "HsCEzbCuaUtUGSCrvwsSbJSlS2HH6TrW0ZeEKUZGTiOH8vPEZxyAEOx974Ku72l4"
+			"token": "HsCEzbCuaUtUGSCrvwsSbJSlS2HH6TrW0ZeEKUZGTiOH8vPEZxyAEOx974Ku72l4",
+			"resolved_modules": ["taurus-boolean", "taurus-control", "taurus-http", "taurus-list", "taurus-number", "taurus-object", "taurus-text"]
 		},
 		{
 			"identifier": "draco-rest",

service.configuration.json

@@ -43,6 +43,8 @@ pub struct ActionServiceConfiguration {
 pub struct RuntimeServiceConfiguration {
     token: String,
     identifier: String,
+    #[serde(default)]
+    resolved_modules: Vec<String>,
 }
 
 #[derive(Serialize, Deserialize, Clone, Default)]
@@ -144,6 +146,24 @@ impl ServiceConfiguration {
         }
     }
 
+    pub fn collect_modules(&self) -> Vec<String> {
+        let actions: Vec<String> = self
+            .actions
+            .iter()
+            .map(|x| format!("action.{}", x.service_name))
+            .collect();
+        let runtime: Vec<String> = self
+            .runtimes
+            .iter()
+            .flat_map(|x| match x.resolved_modules.is_empty() {
+                true => vec![x.identifier.clone()],
+                false => x.resolved_modules.clone(),
+            })
+            .collect();
+
+        vec![actions, runtime].concat()
+    }
+
     pub fn from_path(path: impl AsRef<Path>) -> Self {
         let mut data = String::new();
 
@@ -203,14 +223,20 @@ mod tests {
                 RuntimeServiceConfiguration {
                     token: String::from("taurus-token"),
                     identifier: String::from("taurus"),
+                    resolved_modules: vec![
+                        String::from("taurus-boolean"),
+                        String::from("taurus-number"),
+                    ],
                 },
                 RuntimeServiceConfiguration {
                     token: String::from("draco-rest-token"),
                     identifier: String::from("draco-rest"),
+                    resolved_modules: vec![],
                 },
                 RuntimeServiceConfiguration {
                     token: String::from("draco-cron-token"),
                     identifier: String::from("draco-cron"),
+                    resolved_modules: vec![],
                 },
             ],
         }
@@ -278,6 +304,22 @@ mod tests {
         assert!(!config.has_service(&String::from("action-token"), &String::from("taurus-x")));
     }
 
+    #[test]
+    fn collect_modules_uses_definition_source_identifiers() {
+        let config = fixture();
+
+        assert_eq!(
+            config.collect_modules(),
+            vec![
+                String::from("action.action-identifier"),
+                String::from("taurus-boolean"),
+                String::from("taurus-number"),
+                String::from("draco-rest"),
+                String::from("draco-cron"),
+            ]
+        );
+    }
+
     #[test]
     fn get_action_configuration_requires_matching_token_and_identifier() {
         let config: ServiceConfiguration = SerializableServiceConfiguration {

src/configuration/service.rs

@@ -1,22 +1,30 @@
+use crate::configuration::service::ServiceConfiguration;
 use crate::{authorization::authorization::get_authorization_metadata, telemetry::errors};
 use std::time::Duration;
 use tonic::transport::Channel;
 use tonic::{Extensions, Request};
 
 pub struct SagittariusModuleServiceClient {
+    service: ServiceConfiguration,
     client: tucana::sagittarius::module_service_client::ModuleServiceClient<Channel>,
     token: String,
     unary_rpc_timeout: Duration,
 }
 
 impl SagittariusModuleServiceClient {
-    pub fn new(channel: Channel, token: String, unary_rpc_timeout: Duration) -> Self {
+    pub fn new(
+        channel: Channel,
+        token: String,
+        unary_rpc_timeout: Duration,
+        service_configuration: ServiceConfiguration,
+    ) -> Self {
         let client = tucana::sagittarius::module_service_client::ModuleServiceClient::new(channel);
 
         Self {
             client,
             token,
             unary_rpc_timeout,
+            service: service_configuration,
         }
     }
 
@@ -40,6 +48,7 @@ impl SagittariusModuleServiceClient {
             Extensions::new(),
             tucana::sagittarius::ModuleUpdateRequest {
                 modules: modules_update_request.modules,
+                available_defintition_soruces: self.service.collect_modules(),
             },
         );
         request.set_timeout(self.unary_rpc_timeout);

src/sagittarius/module_service_client_impl.rs

@@ -93,6 +93,7 @@ impl AquilaDynamicServer {
             self.channel.clone(),
             self.token.clone(),
             self.sagittarius_unary_rpc_timeout,
+            self.service_configuration.clone(),
         )));
 
         info!("ModuleService started");