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/src/processor.rs | 26 +++++++++++++++++++-------
 1 file changed, 19 insertions(+), 7 deletions(-)

(limited to 'crates/router/src/processor.rs')

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(())
 }
-- 
cgit v1.2.3