chore(daemon): added validation for empty events (#186)

demeter-run · Nov 4, 2024 · 3cf6914 · 3cf6914
1 parent 6b388c5
commit 3cf6914
Showing 1 changed file with 23 additions and 23 deletions.
diff --git a/src/drivers/monitor/mod.rs b/src/drivers/monitor/mod.rs
@@ -30,32 +30,32 @@ pub async fn subscribe(config: MonitorConfig) -> Result<()> {
             Err(error) => error!(?error, "kafka subscribe error"),
             Ok(message) => {
                 let message = message.borrow();
+                match message.try_into() {
+                    Ok(event) => {
+                        let payload: serde_json::Value =
+                            serde_json::from_slice(message.payload().unwrap())?;
 
-                let payload: serde_json::Value =
-                    serde_json::from_slice(message.payload().unwrap_or_default())?;
-                match payload.get("annotation") {
-                    Some(annotation) => match annotation.get("source") {
-                        Some(v) => {
-                            let source = v.to_string();
-                            if !source_regex.is_match(&source) {
-                                info!(?source, "bypass event. Event source not allowed");
+                        match payload.get("annotation") {
+                            Some(annotation) => match annotation.get("source") {
+                                Some(v) => {
+                                    let source = v.to_string();
+                                    if !source_regex.is_match(&source) {
+                                        info!(?source, "bypass event. Event source not allowed");
+                                        continue;
+                                    }
+                                }
+                                None => {
+                                    info!("bypass event. Event doesnt have a source");
+                                    continue;
+                                }
+                            },
+                            None => {
+                                info!("bypass event. Event doesnt have a source");
                                 continue;
                             }
-                        }
-                        None => {
-                            info!("bypass event. Event doesnt have a source");
-                            continue;
-                        }
-                    },
-                    None => {
-                        info!("bypass event. Event doesnt have a source");
-                        continue;
-                    }
-                };
+                        };
 
-                match message.try_into() {
-                    Ok(event) => {
-                        let event_appliclation = {
+                        let result = {
                             match &event {
                                 Event::ProjectCreated(evt) => {
                                     project::cluster::apply_manifest(cluster.clone(), evt.clone())
@@ -84,7 +84,7 @@ pub async fn subscribe(config: MonitorConfig) -> Result<()> {
                             }
                         };
 
-                        match event_appliclation {
+                        match result {
                             Ok(()) => {
                                 info!(event = event.key(), "Successfully handled event")
                             }