From 1968002d656383069a386bd874c9f0cc83e3116e Mon Sep 17 00:00:00 2001
From: rtkay123 <dev@kanjala.com>
Date: Fri, 15 Aug 2025 19:36:22 +0200
Subject: feat(rule-exec): receive messages

---
 crates/router/Cargo.toml            |  3 ++-
 crates/router/src/processor.rs      | 26 +++++++++++++++++++-------
 crates/router/src/processor/grpc.rs | 27 ---------------------------
 crates/router/src/state.rs          |  6 ++----
 4 files changed, 23 insertions(+), 39 deletions(-)
 delete mode 100644 crates/router/src/processor/grpc.rs

(limited to 'crates/router')

diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 0b7b232..08ccf71 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -14,7 +14,7 @@ bytes = "1.10.1"
 clap = { workspace = true, features = ["derive"] }
 config = { workspace = true, features = ["convert-case", "toml"] }
 futures-util.workspace = true
-moka = { version = "0.12.10", features = ["future"] }
+moka = { workspace = true, features = ["future"] }
 opentelemetry.workspace = true
 opentelemetry-semantic-conventions.workspace = true
 prost.workspace = true
@@ -34,6 +34,7 @@ warden-core = { workspace = true, features = [
     "configuration",
     "serde-time"
 ] }
+warden-middleware.workspace = true
 
 [dependencies.warden-stack]
 workspace = true
diff --git a/crates/router/src/processor.rs b/crates/router/src/processor.rs
index 9afe726..c593393 100644
--- a/crates/router/src/processor.rs
+++ b/crates/router/src/processor.rs
@@ -1,4 +1,3 @@
-pub mod grpc;
 mod load;
 mod publish;
 mod reload;
@@ -11,7 +10,7 @@ use async_nats::jetstream::{
     Context,
     consumer::{Consumer, pull},
 };
-use futures_util::StreamExt;
+use futures_util::{StreamExt, future};
 use tokio::signal;
 use tracing::{error, trace};
 use warden_stack::{Configuration, tracing::SdkTracerProvider};
@@ -46,11 +45,24 @@ async fn run(state: AppHandle) -> anyhow::Result<()> {
     let consumer = consumer?;
 
     // Consume messages from the consumer
-    while let Some(Ok(message)) = consumer.messages().await?.next().await {
-        if let Err(e) = route::route(message, Arc::clone(&state)).await {
-            error!("{}", e.to_string());
-        }
-    }
+
+    let limit = None;
+
+    consumer
+        .messages()
+        .await?
+        .for_each_concurrent(limit, |message| {
+            let state = Arc::clone(&state);
+            tokio::spawn(async move {
+                if let Ok(message) = message
+                    && let Err(e) = route::route(message, Arc::clone(&state)).await
+                {
+                    error!("{}", e.to_string());
+                }
+            });
+            future::ready(())
+        })
+        .await;
 
     Ok(())
 }
diff --git a/crates/router/src/processor/grpc.rs b/crates/router/src/processor/grpc.rs
deleted file mode 100644
index 344f2a1..0000000
--- a/crates/router/src/processor/grpc.rs
+++ /dev/null
@@ -1,27 +0,0 @@
-pub mod interceptor {
-    use opentelemetry::global;
-    use tonic::{
-        Status,
-        service::{Interceptor, interceptor::InterceptedService},
-        transport::Channel,
-    };
-    use tracing::Span;
-    use tracing_opentelemetry::OpenTelemetrySpanExt;
-    use warden_stack::tracing::telemetry::tonic::injector;
-
-    pub type Intercepted = InterceptedService<Channel, MyInterceptor>;
-
-    #[derive(Clone, Copy)]
-    pub struct MyInterceptor;
-
-    impl Interceptor for MyInterceptor {
-        fn call(&mut self, mut request: tonic::Request<()>) -> Result<tonic::Request<()>, Status> {
-            let cx = Span::current().context();
-            global::get_text_map_propagator(|propagator| {
-                propagator.inject_context(&cx, &mut injector::MetadataMap(request.metadata_mut()))
-            });
-
-            Ok(request)
-        }
-    }
-}
diff --git a/crates/router/src/state.rs b/crates/router/src/state.rs
index 4ede2de..a0d9228 100644
--- a/crates/router/src/state.rs
+++ b/crates/router/src/state.rs
@@ -8,12 +8,10 @@ use tracing::error;
 use warden_core::configuration::routing::{
     RoutingConfiguration, query_routing_client::QueryRoutingClient,
 };
+use warden_middleware::grpc::interceptor::{Intercepted, MyInterceptor};
 use warden_stack::Configuration;
 
-use crate::{
-    cnfg::LocalConfig,
-    processor::grpc::interceptor::{Intercepted, MyInterceptor},
-};
+use crate::cnfg::LocalConfig;
 
 #[derive(Clone)]
 pub struct Services {
-- 
cgit v1.2.3