From 0958a318e2b8eedd73649600dbf9150ecfe596fb Mon Sep 17 00:00:00 2001
From: Nastasia Emelianova <e-nastasia@Nastasias-MacBook-Pro.local>
Date: Tue, 9 Jan 2024 15:41:41 -0500
Subject: [PATCH] propagate errors from write_with_retry

---
 src/worker.rs | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/src/worker.rs b/src/worker.rs
index 7aa76f2..dc33027 100644
--- a/src/worker.rs
+++ b/src/worker.rs
@@ -4,7 +4,7 @@ use rmp_serde::Serializer;
 use serde::{ser::SerializeMap, Deserialize, Serialize};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
-    sync::broadcast::{error::RecvError, Receiver},
+    sync::broadcast::{error::{RecvError, self}, Receiver},
     time::Duration,
 };
 
@@ -116,9 +116,17 @@ where
                     match self.write_with_retry(&record).await {
                         Ok(_) => {}
                         Err(e) => {
-                            error!("GOT ERROR FROM write_with_retry: {:#?}", e.to_string());
-                            // TODO: attempt to reconnect the tcp stream we're writing to
-                            continue;
+                            match e {
+                                Error::MaxRetriesExceeded => {
+                                    error!("Reached MaxRetriesExceeded");
+                                    break;
+                                },
+                                Error::ConnectionClosed => {
+                                    error!("Reached ConnectionClosed");
+                                    break;
+                                },
+                                _ => continue,
+                            }
                         }
                     };
                 }