BroadcastSinkError

pragmaxim-com · Jul 1, 2024 · dfd267f · dfd267f
1 parent 9433759
commit dfd267f
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 7 deletions.
diff --git a/benches/broadcast_sink_benchmark.rs b/benches/broadcast_sink_benchmark.rs
@@ -1,4 +1,4 @@
-use broadcast_sink::{Consumer, StreamBroadcastSinkExt};
+use broadcast_sink::{BroadcastSinkError, Consumer, StreamBroadcastSinkExt};
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
 use futures::stream::{self, Stream};
 use std::sync::{Arc, RwLock};
@@ -21,7 +21,7 @@ impl MultiplyX {
 }
 
 impl Consumer<u64> for MultiplyX {
-    fn consume(&mut self, _: &u64) -> Result<(), &'static str> {
+    fn consume(&mut self, _: &u64) -> Result<(), BroadcastSinkError> {
         let mut x = self.state.x.write().unwrap();
         *x *= 5;
         println!("Consumer 1 processed item");
@@ -40,7 +40,7 @@ impl MultiplyY {
 }
 
 impl Consumer<u64> for MultiplyY {
-    fn consume(&mut self, _: &u64) -> Result<(), &'static str> {
+    fn consume(&mut self, _: &u64) -> Result<(), BroadcastSinkError> {
         let mut y = self.state.y.write().unwrap();
         *y *= 10;
         println!("Consumer 2 processed item");

diff --git a/src/lib.rs b/src/lib.rs
@@ -20,8 +20,13 @@ use tokio::sync::{broadcast, Barrier, Mutex};
 use tokio::task;
 use tokio_stream::wrappers::BroadcastStream;
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BroadcastSinkError {
+    message: String,
+}
+
 pub trait Consumer<T>: Send + Sync {
-    fn consume(&mut self, item: &T) -> Result<(), &'static str>;
+    fn consume(&mut self, item: &T) -> Result<(), BroadcastSinkError>;
 }
 
 pin_project! {
@@ -63,7 +68,7 @@ where
                 while let Some(Ok(item)) = stream.next().await {
                     let mut consumer = consumer.lock().await;
                     if let Err(e) = consumer.consume(&item) {
-                        error!("BroadcastSink consumer error occurred: {:?}", e);
+                        error!("BroadcastSink consumer error occurred: {:?}", e.message);
                     }
                     barrier_clone.wait().await;
                     active_count_clone.fetch_sub(1, Ordering::SeqCst);
@@ -148,7 +153,7 @@ mod tests {
     }
 
     impl Consumer<u64> for MultiplyX {
-        fn consume(&mut self, _: &u64) -> Result<(), &'static str> {
+        fn consume(&mut self, _: &u64) -> Result<(), BroadcastSinkError> {
             let mut x = self.state.x.write().unwrap();
             *x *= 5;
             println!("Consumer X processed item");
@@ -167,7 +172,7 @@ mod tests {
     }
 
     impl Consumer<u64> for MultiplyY {
-        fn consume(&mut self, _: &u64) -> Result<(), &'static str> {
+        fn consume(&mut self, _: &u64) -> Result<(), BroadcastSinkError> {
             let mut y = self.state.y.write().unwrap();
             *y *= 10;
             println!("Consumer Y processed item");