fix: made config same to Aquila and Taurus

Author: raphael-goetz

Cargo.lock

@@ -1,7 +1,7 @@
 use base::{
     extract_flow_setting_field,
     runner::{ServerContext, ServerRunner},
-    store::FlowIdenfiyResult,
+    store::FlowIdentifyResult,
     traits::{IdentifiableFlow, LoadConfig, Server as ServerTrait},
 };
 use code0_flow::flow_config::env_with_default;
@@ -64,7 +64,7 @@ impl ServerTrait<HttpServerConfig> for HttpServer {
                         };
 
                         match store.get_possible_flow_match(pattern, route).await {
-                            FlowIdenfiyResult::Single(flow) => {
+                            FlowIdentifyResult::Single(flow) => {
                                 execute_flow(flow, request, store).await
                             }
                             _ => Some(HttpResponse::internal_server_error(

adapter/rest/src/main.rs

@@ -15,3 +15,4 @@ tonic-health = { workspace = true }
 uuid = { workspace = true }
 prost = { workspace = true }
 futures-lite = { workspace = true }
+log = { workspace = true }

crates/base/Cargo.toml

@@ -32,6 +32,11 @@ pub struct AdapterConfig {
     /// Port on which the adapter's Health Service server will listen.
     pub grpc_port: u16,
 
+    /// GRPC Host
+    ///
+    /// Host on which the adapter's Health Service server will listen.
+    pub grpc_host: String,
+
     /// Aquila URL
     ///
     /// URL of the Aquila server to connect to.
@@ -45,7 +50,7 @@ pub struct AdapterConfig {
     /// Is Monitored
     ///
     /// If true the Adapter will expose a grpc health service server.
-    pub is_monitored: bool,
+    pub with_health_service: bool,
 }
 
 impl AdapterConfig {
@@ -57,6 +62,7 @@ impl AdapterConfig {
         let nats_bucket =
             code0_flow::flow_config::env_with_default("NATS_BUCKET", String::from("flow_store"));
         let grpc_port = code0_flow::flow_config::env_with_default("GRPC_PORT", 50051);
+        let grpc_host = code0_flow::flow_config::env_with_default("GRPC_HOST", String::from("localhost"));
         let aquila_url = code0_flow::flow_config::env_with_default(
             "AQUILA_URL",
             String::from("grpc://localhost:50051"),
@@ -69,17 +75,18 @@ impl AdapterConfig {
             "DEFINITION_PATH",
             String::from("./definition.yaml"),
         );
-        let is_monitored = code0_flow::flow_config::env_with_default("IS_MONITORED", false);
+        let with_health_service = code0_flow::flow_config::env_with_default("WITH_HEALTH_SERVICE", false);
 
         Self {
             environment,
             nats_bucket,
             mode,
             nats_url,
             grpc_port,
+            grpc_host,
             aquila_url,
             definition_path,
-            is_monitored,
+            with_health_service,
         }
     }
 

crates/base/src/config.rs

@@ -61,10 +61,10 @@ impl<C: LoadConfig> ServerRunner<C> {
             definition_service.send().await;
         }
 
-        if config.is_monitored {
+        if config.with_health_service {
             let health_service =
                 code0_flow::flow_health::HealthService::new(config.nats_url.clone());
-            let address = format!("127.0.0.1:{}", config.grpc_port).parse()?;
+            let address = format!("{}:{}", config.grpc_host, config.grpc_port).parse()?;
 
             tokio::spawn(async move {
                 let _ = Server::builder()
@@ -73,7 +73,7 @@ impl<C: LoadConfig> ServerRunner<C> {
                     .await;
             });
 
-            println!("Health server started at 127.0.0.1:{}", config.grpc_port);
+            log::info!("Health server started at {}:{}", config.grpc_host, config.grpc_port);
         }
 
         self.server.init(&self.context).await?;

crates/base/src/runner.rs

@@ -10,7 +10,7 @@ pub struct AdapterStore {
     kv: async_nats::jetstream::kv::Store,
 }
 
-pub enum FlowIdenfiyResult {
+pub enum FlowIdentifyResult {
     None,
     Single(ValidationFlow),
     Multiple(Vec<ValidationFlow>),
@@ -20,19 +20,24 @@ impl AdapterStore {
     pub async fn from_url(url: String, bucket: String) -> Self {
         let client = match async_nats::connect(url).await {
             Ok(client) => client,
-            Err(err) => panic!("Failed to connect to NATS server: {}", err),
+            Err(err) => panic!("Failed to connect to NATS server: {:?}", err),
         };
 
-        let jetstream = async_nats::jetstream::new(client.clone());
+        let stream = async_nats::jetstream::new(client.clone());
 
-        let _ = jetstream
+        match stream
             .create_key_value(Config {
                 bucket: bucket.clone(),
                 ..Default::default()
             })
-            .await;
+            .await {
+            Ok(_) => {
+                log::info!("Successfully created NATS bucket");
+            },
+            Err(err) => panic!("Failed to create NATS bucket: {:?}", err),
+        }
 
-        let kv = match jetstream.get_key_value(bucket).await {
+        let kv = match stream.get_key_value(bucket).await {
             Ok(kv) => kv,
             Err(err) => panic!("Failed to get key-value store: {}", err),
         };
@@ -63,21 +68,19 @@ impl AdapterStore {
         &self,
         pattern: String,
         id: I,
-    ) -> FlowIdenfiyResult {
+    ) -> FlowIdentifyResult {
         let mut collector = Vec::new();
         let mut keys = match self.kv.keys().await {
             Ok(keys) => keys.boxed(),
             Err(err) => {
-                eprintln!("Failed to get keys: {}", err);
-                return FlowIdenfiyResult::None;
+                log::error!("Failed to get keys: {}", err);
+                return FlowIdentifyResult::None;
             }
         };
 
         while let Ok(Some(key)) = keys.try_next().await {
-            println!("comparing: key: {} pattern {:?}", key, pattern);
 
             if !Self::is_matching_key(&pattern, &key) {
-                println!("Key does not match pattern: {}", key);
                 continue;
             }
 
@@ -92,9 +95,9 @@ impl AdapterStore {
         }
 
         match collector.len() {
-            0 => FlowIdenfiyResult::None,
-            1 => FlowIdenfiyResult::Single(collector[0].clone()),
-            _ => FlowIdenfiyResult::Multiple(collector),
+            0 => FlowIdentifyResult::None,
+            1 => FlowIdentifyResult::Single(collector[0].clone()),
+            _ => FlowIdentifyResult::Multiple(collector),
         }
     }
 
@@ -130,16 +133,15 @@ impl AdapterStore {
         match result {
             Ok(message) => match Value::decode(message.payload) {
                 Ok(value) => {
-                    println!("Response: {:?}", &value);
                     Some(value)
                 }
                 Err(err) => {
-                    eprintln!("Failed to decode response from NATS server: {}", err);
-                    return None;
+                    log::error!("Failed to decode response from NATS server: {:?}", err);
+                    None
                 }
             },
             Err(err) => {
-                eprintln!("Failed to send request to NATS server: {}", err);
+                log::error!("Failed to send request to NATS server: {:?}", err);
                 None
             }
         }
@@ -154,22 +156,19 @@ impl AdapterStore {
     }
 
     fn is_matching_key(pattern: &String, key: &String) -> bool {
-        let splitted_pattern = pattern.split(".");
-        let splitted_key = key.split(".").collect::<Vec<&str>>();
-
-        let zip = splitted_pattern.into_iter().zip(splitted_key);
+        let split_pattern = pattern.split(".");
+        let split_key = key.split(".").collect::<Vec<&str>>();
+        let zip = split_pattern.into_iter().zip(split_key);
 
         for (pattern_part, key_part) in zip {
             if pattern_part == "*" {
                 continue;
             }
 
             if pattern_part != key_part {
-                println!("matching: pattern: {} key: {}", pattern_part, key_part);
                 return false;
             }
         }
-        println!("pattern was correct");
         true
     }
 }