feat: send_buffered function to send events (#1829)

* fix: send_event error handling
* enha: handle_delivery_result function

Author: carneiro-cw

Cargo.lock

@@ -68,6 +68,7 @@ crossbeam-channel = "=0.5.13"
 futures = "=0.3.30"
 futures-timer = "=3.0.3"
 futures-util = "=0.3.31"
+futures-channel = "=0.3.31"
 
 # ethereum / rpc
 ethabi = "=18.0.0"

Cargo.toml

@@ -2,8 +2,11 @@ use anyhow::Result;
 use clap::Parser;
 use clap::ValueEnum;
 use display_json::DebugAsJson;
+use futures::Stream;
+use futures::StreamExt;
 use rdkafka::message::Header;
 use rdkafka::message::OwnedHeaders;
+use rdkafka::producer::future_producer::OwnedDeliveryResult;
 use rdkafka::producer::DeliveryFuture;
 use rdkafka::producer::FutureProducer;
 use rdkafka::producer::FutureRecord;
@@ -161,14 +164,19 @@ impl KafkaConnector {
     }
 
     pub async fn send_event<T: Event>(&self, event: T) -> Result<()> {
-        match self.queue_event(event) {
-            Ok(fut) =>
-                if let Err(e) = fut.await {
-                    log_and_err!(reason = e, "failed to publish kafka event")
-                } else {
-                    Ok(())
-                },
-            Err(e) => Err(e),
-        }
+        handle_delivery_result(self.queue_event(event)?.await)
+    }
+
+    pub fn send_buffered<T: Event>(&self, events: Vec<T>, buffer_size: usize) -> Result<impl Stream<Item = Result<()>>> {
+        let futures: Vec<DeliveryFuture> = events.into_iter().map(|event| self.queue_event(event)).collect::<Result<Vec<_>, _>>()?; // This could fail because the queue is full
+        Ok(futures::stream::iter(futures).buffered(buffer_size).map(handle_delivery_result))
+    }
+}
+
+fn handle_delivery_result(res: Result<OwnedDeliveryResult, futures_channel::oneshot::Canceled>) -> Result<()> {
+    match res {
+        Err(e) => log_and_err!(reason = e, "failed to publish kafka event"),
+        Ok(Err((e, _))) => log_and_err!(reason = e, "failed to publish kafka event"),
+        Ok(_) => Ok(()),
     }
 }