Merge pull request #239 from code0-tech/deps/code0-flow

deps/code0-flow

Author: raphael-goetz

Cargo.lock

@@ -7,8 +7,8 @@ version = "0.0.0"
 edition = "2024"
 
 [workspace.dependencies]
-code0-flow = { version = "0.0.33" }
-tucana = { version = "0.0.73", features = ["aquila"] }
+code0-flow = { version = "0.0.36" }
+tucana = { version = "0.0.74", features = ["aquila"] }
 serde_json = { version = "1.0.138" }
 log = "0.4.27"
 env_logger = "0.11.8"

Cargo.toml

@@ -3,6 +3,7 @@ use base::{
     store::{FlowExecutionResult, FlowIdentifyResult},
     traits::Server as ServerTrait,
 };
+use code0_flow::flow_service::ModuleDefinitionAppendix;
 use http_body_util::{BodyExt, Full};
 use hyper::server::conn::http1;
 use hyper::{Request, Response};
@@ -17,7 +18,7 @@ use std::sync::Arc;
 use tokio::net::TcpListener;
 use tonic::async_trait;
 use tucana::shared::{
-    AdapterStatusConfiguration, Struct, ValidationFlow, Value, helper::value::ToValue, value::Kind,
+    Endpoint, ModuleDefinition, Struct, ValidationFlow, Value, helper::value::ToValue, value::Kind,
 };
 
 use crate::response::{error_to_http_response, value_to_http_response};
@@ -42,14 +43,18 @@ async fn main() {
     let addr = runner.get_server_config().port;
     let host = runner.get_server_config().host.clone();
 
-    let configs = vec![AdapterStatusConfiguration {
-        flow_type_identifiers: vec![String::from("REST")],
-        data: Some(
-            tucana::shared::adapter_status_configuration::Data::Endpoint(format!(
-                r"{}:{}/${{project_slug}}/${{flow_setting_identifier}}",
-                host, addr
+    let configs = vec![ModuleDefinitionAppendix {
+        module_identifier: String::from("draco-rest"),
+        definitions: vec![ModuleDefinition {
+            flow_type_identifier: vec![String::from("REST")],
+            value: Some(tucana::shared::module_definition::Value::Endpoint(
+                Endpoint {
+                    host,
+                    port: addr as i64,
+                    endpoint: String::from(r"/${{project_slug}}${{httpURL}}"),
+                },
             )),
-        ),
+        }],
     }];
     match runner.serve(configs).await {
         Ok(_) => (),

adapter/rest/src/main.rs

@@ -9,15 +9,13 @@ use tonic::{
 use tucana::{
     aquila::{
         RuntimeStatusUpdateRequest, runtime_status_service_client::RuntimeStatusServiceClient,
-        runtime_status_update_request::Status,
     },
-    shared::{AdapterRuntimeStatus, AdapterStatusConfiguration},
+    shared::ModuleStatus,
 };
 
 pub struct DracoRuntimeStatusService {
     channel: Channel,
     identifier: String,
-    configs: Vec<AdapterStatusConfiguration>,
     aquila_token: String,
 }
 
@@ -69,33 +67,22 @@ pub async fn create_channel_with_retry(channel_name: &str, url: String) -> Chann
     }
 }
 impl DracoRuntimeStatusService {
-    pub async fn from_url(
-        aquila_url: String,
-        aquila_token: String,
-        identifier: String,
-        configs: Vec<AdapterStatusConfiguration>,
-    ) -> Self {
+    pub async fn from_url(aquila_url: String, aquila_token: String, identifier: String) -> Self {
         let channel = create_channel_with_retry("Aquila", aquila_url).await;
-        Self::new(channel, identifier, configs, aquila_token)
+        Self::new(channel, identifier, aquila_token)
     }
 
-    pub fn new(
-        channel: Channel,
-        identifier: String,
-        configs: Vec<AdapterStatusConfiguration>,
-        aquila_token: String,
-    ) -> Self {
+    pub fn new(channel: Channel, identifier: String, aquila_token: String) -> Self {
         DracoRuntimeStatusService {
             channel,
             identifier,
-            configs,
             aquila_token,
         }
     }
 
     pub async fn update_runtime_status_by_status(
         &self,
-        status: tucana::shared::adapter_runtime_status::Status,
+        status: tucana::shared::module_status::StatusVariant,
     ) {
         log::info!("Updating the current runtime status!");
         let mut client = RuntimeStatusServiceClient::new(self.channel.clone());
@@ -113,12 +100,11 @@ impl DracoRuntimeStatusService {
             get_authorization_metadata(&self.aquila_token),
             Extensions::new(),
             RuntimeStatusUpdateRequest {
-                status: Some(Status::AdapterRuntimeStatus(AdapterRuntimeStatus {
+                status: Some(ModuleStatus {
                     status: status.into(),
                     timestamp: timestamp as i64,
                     identifier: self.identifier.clone(),
-                    configurations: self.configs.clone(),
-                })),
+                }),
             },
         );
 

crates/base/src/client/mod.rs

@@ -4,12 +4,11 @@ use crate::{
     store::AdapterStore,
     traits::{LoadConfig, Server as AdapterServer},
 };
-use code0_flow::flow_service::FlowUpdateService;
+use code0_flow::flow_service::{FlowUpdateService, ModuleDefinitionAppendix};
 use std::{sync::Arc, time::Duration};
 use tokio::{signal, task::JoinHandle, time::sleep};
 use tonic::transport::Server;
 use tonic_health::pb::health_server::HealthServer;
-use tucana::shared::AdapterStatusConfiguration;
 
 /// Context passed to adapter server implementations containing all shared resources
 pub struct ServerContext<C: LoadConfig> {
@@ -56,10 +55,7 @@ impl<C: LoadConfig> ServerRunner<C> {
         })
     }
 
-    pub async fn serve(
-        self,
-        runtime_config: Vec<AdapterStatusConfiguration>,
-    ) -> anyhow::Result<()> {
+    pub async fn serve(self, appendix: Vec<ModuleDefinitionAppendix>) -> anyhow::Result<()> {
         let config = self.context.adapter_config.clone();
         let mut runtime_status_service: Option<Arc<DracoRuntimeStatusService>> = None;
         let mut runtime_status_heartbeat_task: Option<JoinHandle<()>> = None;
@@ -71,26 +67,27 @@ impl<C: LoadConfig> ServerRunner<C> {
                     config.aquila_url.clone(),
                     config.aquila_token.clone(),
                     config.draco_variant.clone(),
-                    runtime_config,
                 )
                 .await,
             ));
 
             if let Some(ser) = &runtime_status_service {
                 ser.update_runtime_status_by_status(
-                    tucana::shared::adapter_runtime_status::Status::NotReady,
+                    tucana::shared::module_status::StatusVariant::NotReady,
                 )
                 .await;
             };
 
             let service_name = format!("draco-{}", config.draco_variant.to_lowercase());
+
             let mut definition_service = FlowUpdateService::from_url(
                 config.aquila_url.clone(),
                 config.definition_path.as_str(),
                 config.aquila_token.clone(),
             )
             .await
-            .with_definition_source(service_name);
+            .with_definition_source(service_name)
+            .with_appendix(appendix);
 
             let mut success = false;
             let mut count = 1;
@@ -144,7 +141,7 @@ impl<C: LoadConfig> ServerRunner<C> {
 
         if let Some(ser) = &runtime_status_service {
             ser.update_runtime_status_by_status(
-                tucana::shared::adapter_runtime_status::Status::Running,
+                tucana::shared::module_status::StatusVariant::Running,
             )
             .await;
 
@@ -163,7 +160,7 @@ impl<C: LoadConfig> ServerRunner<C> {
                         interval.tick().await;
                         status_service
                             .update_runtime_status_by_status(
-                                tucana::shared::adapter_runtime_status::Status::Running,
+                                tucana::shared::module_status::StatusVariant::Running,
                             )
                             .await;
                     }
@@ -251,7 +248,7 @@ impl<C: LoadConfig> ServerRunner<C> {
 
         if let Some(ser) = &runtime_status_service {
             ser.update_runtime_status_by_status(
-                tucana::shared::adapter_runtime_status::Status::Stopped,
+                tucana::shared::module_status::StatusVariant::Stopped,
             )
             .await;
         };