code-check-pipe

.env-example

@@ -1,6 +1,9 @@
 ENVIRONMENT='development'
-MODE='dynamic'
-REDIS_URL='redis://localhost:6379'
-RABBITMQ_URL='amqp://localhost:5672'
+MODE='static'
+NATS_URL='nats://localhost:4222'
+NATS_BUCKET='flow_store'
+GRPC_HOST='127.0.0.1'
+GRPC_PORT=8081
+WITH_HEALTH_SERVICE=false
 AQUILA_URL='http://localhost:8080'
-PORT=8080
+DEFINITION_PATH='./definitions'

.github/workflows/ci.yml

@@ -0,0 +1,104 @@
+name: Rust CI/CD Pipeline
+
+on:
+  push:
+  pull_request:
+    types: [opened, reopened]
+
+env:
+  RUST_BACKTRACE: 1
+
+jobs:
+  # Job 1: Format Check
+  format:
+    name: Format Check
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v5
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          components: rustfmt
+
+      - name: Check formatting
+        run: cargo fmt --all -- --check
+
+  # Job 2: Lint Check
+  lint:
+    name: Lint Check
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v5
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          components: clippy
+
+      - name: Cache cargo registry
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            target
+          key: ${{ runner.os }}-cargo-lint-${{ hashFiles('**/Cargo.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-cargo-
+
+      - name: Run clippy
+        run: cargo clippy --all-targets --all-features
+
+  # Job 3: Build
+  build:
+    name: Build
+    needs: [ lint, format ]
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v5
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Cache cargo registry
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            target
+          key: ${{ runner.os }}-cargo-build-${{ hashFiles('**/Cargo.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-cargo-
+
+      - name: Build project
+        run: cargo build --verbose --all-features
+
+  test:
+    name: Test
+    needs: build
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v5
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Cache cargo registry
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            target
+          key: ${{ runner.os }}-cargo-build-${{ hashFiles('**/Cargo.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-cargo-
+
+      - name: Run Tests
+        run:  cargo test --verbose --all-features

adapter/rest/src/.env

@@ -1,5 +1,11 @@
-HTTP_PORT=8081
-REDIS_URL=redis://localhost:6379
-RABBITMQ_URL=amqp://localhost:5672
-AQUILA_URL=http://localhost:8080
-IS_STATIC=false
+HTTP_SERVER_PORT=8080
+
+ENVIRONMENT='development'
+MODE='static'
+NATS_URL='nats://localhost:4222'
+NATS_BUCKET='flow_store'
+GRPC_HOST='127.0.0.1'
+GRPC_PORT=8081
+WITH_HEALTH_SERVICE=false
+AQUILA_URL='http://localhost:8080'
+DEFINITION_PATH='./definitions'

adapter/rest/src/main.rs

@@ -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(

crates/base/Cargo.toml

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

crates/base/src/config.rs

@@ -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,8 @@ 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"),
@@ -67,19 +74,21 @@ impl AdapterConfig {
         let mode = code0_flow::flow_config::env_with_default("MODE", Mode::STATIC);
         let definition_path = code0_flow::flow_config::env_with_default(
             "DEFINITION_PATH",
-            String::from("./definition.yaml"),
+            String::from("./definition"),
         );
-        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/lib.rs

@@ -17,10 +17,10 @@ pub fn extract_flow_setting_field(
 
         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());
-                }
+            if k == field_name
+                && let Some(Kind::StringValue(s)) = &v.kind
+            {
+                return Some(s.clone());
             }
             None
         })

crates/base/src/runner.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,11 @@ 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?;