feat(rule-exec): receive messages

3 files changed, 21 insertions, 38 deletions

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 {