feat: added group id consumer

src/bin/daemon.rs

@@ -2,6 +2,7 @@ use std::env;
 
 use anyhow::Result;
 use dotenv::dotenv;
+use fabric::drivers::monitor::MonitorConfig;
 use serde::Deserialize;
 use tracing::Level;
 use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
@@ -22,12 +23,13 @@ async fn main() -> Result<()> {
 
     let config = Config::new()?;
 
-    fabric::drivers::monitor::subscribe(&config.brokers).await
+    fabric::drivers::monitor::subscribe(config.into()).await
 }
 
 #[derive(Debug, Deserialize)]
 struct Config {
     brokers: String,
+    consumer_name: String,
 }
 impl Config {
     pub fn new() -> Result<Self> {
@@ -43,3 +45,12 @@ impl Config {
         Ok(config)
     }
 }
+
+impl From<Config> for MonitorConfig {
+    fn from(value: Config) -> Self {
+        Self {
+            brokers: value.brokers,
+            consumer_name: value.consumer_name,
+        }
+    }
+}

src/drivers/monitor/mod.rs

@@ -14,14 +14,14 @@ use crate::{
     driven::k8s::K8sCluster,
 };
 
-pub async fn subscribe(brokers: &str) -> Result<()> {
+pub async fn subscribe(config: MonitorConfig) -> Result<()> {
     let k8s_cluster = Arc::new(K8sCluster::new().await?);
 
     let topic = String::from("events");
 
     let consumer: StreamConsumer = ClientConfig::new()
-        .set("bootstrap.servers", brokers)
-        .set("group.id", "clusters")
+        .set("bootstrap.servers", &config.brokers)
+        .set("group.id", &config.consumer_name)
         .create()?;
 
     consumer.subscribe(&[&topic])?;
@@ -50,3 +50,8 @@ pub async fn subscribe(brokers: &str) -> Result<()> {
         };
     }
 }
+
+pub struct MonitorConfig {
+    pub brokers: String,
+    pub consumer_name: String,
+}