feat: made use of new code0-flow feature

Author: raphael-goetz

adapter/rest/src/.env

@@ -2,3 +2,4 @@ PORT=8081
 REDIS_URL=redis://localhost:6379
 RABBITMQ_URL=amqp://localhost:5672
 AQUILA_URL=http://localhost:8080
+IS_STATIC=true

adapter/rest/src/main.rs

@@ -3,7 +3,11 @@ pub mod store;
 mod types;
 
 use code0_flow::{
-    flow_queue::service::RabbitmqClient, flow_store::connection::create_flow_store_connection,
+    flow_queue::service::RabbitmqClient,
+    flow_store::{
+        connection::create_flow_store_connection,
+        service::{FlowStoreService, FlowStoreServiceBase},
+    },
 };
 use config::FromEnv;
 use http::{
@@ -13,19 +17,22 @@ use http::{
 };
 use queue::queue::handle_connection;
 use std::{future::Future, pin::Pin, sync::Arc};
+use tokio::sync::Mutex;
 use types::{get_data_types, get_flow_types};
 
 pub struct FlowConnectionHandler {
-    flow_store: code0_flow::flow_store::connection::FlowStore,
+    flow_store: Arc<Mutex<FlowStoreService>>,
     rabbitmq_client: Arc<RabbitmqClient>,
 }
 
 impl FlowConnectionHandler {
     pub async fn new(config: &Config) -> Self {
         let flow_store = create_flow_store_connection(config.redis_url.clone()).await;
+        let flow_store_service = Arc::new(Mutex::new(FlowStoreServiceBase::new(flow_store).await));
+
         let rabbitmq_client = Arc::new(RabbitmqClient::new(config.rabbitmq_url.as_str()).await);
         FlowConnectionHandler {
-            flow_store,
+            flow_store: flow_store_service,
             rabbitmq_client,
         }
     }

adapter/rest/src/queue/mod.rs

@@ -2,10 +2,11 @@ pub mod queue {
 
     use code0_flow::{
         flow_queue::service::{Message, RabbitmqClient},
-        flow_store::connection::FlowStore,
+        flow_store::service::FlowStoreService,
     };
     use http::{request::HttpRequest, response::HttpResponse};
     use std::{collections::HashMap, sync::Arc, time::Duration};
+    use tokio::sync::Mutex;
     use tucana::shared::{Struct, Value};
     use validator::{resolver::flow_resolver::resolve_flow, verify_flow};
 
@@ -30,11 +31,11 @@ pub mod queue {
 
     pub async fn handle_connection(
         mut request: HttpRequest,
-        flow_store: FlowStore,
+        flow_store: Arc<Mutex<FlowStoreService>>,
         rabbitmq_client: Arc<RabbitmqClient>,
     ) -> Option<HttpResponse> {
         // Check if a flow exists for the given settings, return none if not exsist for http handler
-        let flow_exists = check_flow_exists(&flow_store, &request).await;
+        let flow_exists = check_flow_exists(flow_store, &request).await;
 
         let flow_result = match flow_exists {
             Some(flow) => flow,

adapter/rest/src/store/mod.rs

@@ -1,110 +1,75 @@
 pub mod store {
-    use code0_flow::flow_store::connection::FlowStore;
+    use std::sync::Arc;
+
+    use code0_flow::flow_store::service::{FlowStoreService, FlowStoreServiceBase};
     use http::request::HttpRequest;
-    use redis::{AsyncCommands, JsonAsyncCommands};
     use regex::Regex;
-    use tucana::shared::{value::Kind, Flow, Struct};
+    use tokio::sync::Mutex;
+    use tucana::shared::{value::Kind, Flow, FlowSetting};
 
     //The regex is required for later purposes --> resolve the parameter of the url
     pub struct FlowExistResult {
         pub flow: Flow,
         pub regex_pattern: Regex,
     }
 
-    pub async fn check_flow_exists(
-        flow_store: &FlowStore,
-        request: &HttpRequest,
-    ) -> Option<FlowExistResult> {
-        let mut store = flow_store.lock().await;
-
-        // Get all keys from Redis
-        let keys: Vec<String> = store.keys("*").await.unwrap_or_default();
-        let mut result: Vec<Flow> = Vec::new();
-
-        // Retrieve JSON values for each key
-        for key in keys {
-            if let Ok(json_value) = store.json_get::<&String, &str, String>(&key, "$").await {
-                let flow = match serde_json::from_str::<Vec<Flow>>(json_value.as_str()) {
-                    Ok(flow) => flow[0].clone(),
-                    Err(_) => continue,
-                };
-
-                result.push(flow);
+    fn extract_field(settings: &[FlowSetting], def_key: &str, field_name: &str) -> Option<String> {
+        settings.iter().find_map(|setting| {
+            let def = setting.definition.as_ref()?;
+            if def.key != def_key {
+                return None;
             }
-        }
-
-        for flow in result {
-            let mut correct_url = false;
-            let mut correct_method = false;
-            let mut flow_regex: Option<Regex> = None;
-
-            for setting in flow.settings.clone() {
-                let definition = match setting.definition {
-                    Some(definition) => definition,
-                    None => continue,
-                };
-
-                if definition.key == "HTTP_METHOD" {
-                    let object: Struct = match setting.object {
-                        Some(object) => object,
-                        None => continue,
-                    };
 
-                    for field in object.fields {
-                        if field.0 == "method" {
-                            if let Some(Kind::StringValue(method)) = field.1.kind {
-                                if method == request.method.to_string() {
-                                    correct_method = true;
-                                }
-                            }
-                        }
+            let obj = setting.object.as_ref()?;
+            obj.fields.iter().find_map(|(k, v)| {
+                if k == field_name {
+                    if let Some(Kind::StringValue(s)) = &v.kind {
+                        return Some(s.clone());
                     }
-
-                    continue;
                 }
+                None
+            })
+        })
+    }
 
-                if definition.key == "URL" {
-                    let object: Struct = match setting.object {
-                        Some(object) => object,
-                        None => continue,
-                    };
+    pub async fn check_flow_exists(
+        flow_store: Arc<Mutex<FlowStoreService>>,
+        request: &HttpRequest,
+    ) -> Option<FlowExistResult> {
+        let flows = {
+            let mut store = flow_store.lock().await;
+            let pattern = format!("*::*::{}::{}", request.host, request.method.to_string());
+            let result = store.query_flows(pattern).await;
+
+            match result {
+                Ok(flows) => flows.flows,
+                Err(_) => return None,
+            }
+        };
 
-                    for field in object.fields {
-                        if field.0 == "url" {
-                            if let Some(Kind::StringValue(regex_str)) = field.1.kind {
-                                let regex = match regex::Regex::new(&regex_str) {
-                                    Ok(regex) => regex,
-                                    Err(err) => {
-                                        log::error!("Failed to compile regex: {}", err);
-                                        continue;
-                                    }
-                                };
+        for flow in flows {
+            let url = extract_field(&flow.settings, "HTTP_URL", "url");
 
-                                if regex.is_match(&request.path) {
-                                    correct_url = true;
-                                    flow_regex = Some(regex);
-                                }
-                            }
-                        }
-                    }
+            let regex_str = match url {
+                Some(string) => string,
+                None => continue,
+            };
 
+            let regex = match regex::Regex::new(&regex_str) {
+                Ok(regex) => regex,
+                Err(err) => {
+                    log::error!("Failed to compile regex: {}", err);
                     continue;
                 }
-            }
-
-            if correct_method && correct_url {
-                let regex_pattern = match flow_regex {
-                    Some(regex) => regex.clone(),
-                    None => continue,
-                };
+            };
 
+            if regex.is_match(&request.path) {
                 return Some(FlowExistResult {
                     flow,
-                    regex_pattern,
+                    regex_pattern: regex,
                 });
             }
         }
-
         None
     }
 }