Make consume_async propagate errors via Result

- consume_async now requires Future<Output = anyhow::Result<()>>
- kdb_write_consumer uses ? operator to propagate connection/insert errors
- Errors are surfaced in teardown via block_on(handle)??
- Updated all async consumers in examples/tests to return Ok(())

Author: 0-jake-0

wingfoil/examples/async/README.md

@@ -52,6 +52,7 @@ let consumer = async move |mut source: Pin<Box<dyn FutStream<u32>>>| {
     while let Some((time, value)) = source.next().await {
         println!("{time:?}, {value:?}");
     }
+    Ok(())
 };
 
 produce_async(producer)

wingfoil/examples/async/main.rs

@@ -11,7 +11,7 @@ fn main() {
     let run_for = RunFor::Duration(period * 5);
     let run_mode = RunMode::RealTime;
 
-    let producer = move |_ctx: RunParams| async move {
+    let producer = move |_| async move {
         Ok(stream! {
             for i in 0.. {
                 tokio::time::sleep(period).await; // simulate waiting IO
@@ -25,6 +25,7 @@ fn main() {
         while let Some((time, value)) = source.next().await {
             println!("{time:?}, {value:?}");
         }
+        Ok(())
     };
 
     produce_async(producer)

wingfoil/examples/rfq/order_gateway.rs

@@ -17,10 +17,13 @@ pub trait OrderGateway {
 pub struct RealTimeOrderGateway {}
 
 impl RealTimeOrderGateway {
-    async fn consume_orders(mut source: Pin<Box<dyn FutStream<TinyVec<[Order; 1]>>>>) {
+    async fn consume_orders(
+        mut source: Pin<Box<dyn FutStream<TinyVec<[Order; 1]>>>>,
+    ) -> anyhow::Result<()> {
         while let Some((_time, _value)) = source.next().await {
             //println!("{time:?}, {value:?}");
         }
+        Ok(())
     }
 }
 

wingfoil/src/adapters/kdb/write.rs

@@ -88,26 +88,20 @@ async fn kdb_write_consumer<T>(
     connection: KdbConnection,
     table_name: String,
     mut source: Pin<Box<dyn FutStream<TinyVec<[T; 1]>>>>,
-) where
+) -> anyhow::Result<()>
+where
     T: Element + Send + KdbSerialize + 'static,
 {
     let creds = connection.credentials_string();
 
     // Connect to KDB
-    let mut socket = match QStream::connect(
+    let mut socket = QStream::connect(
         ConnectionMethod::TCP,
         &connection.host,
         connection.port,
         &creds,
     )
-    .await
-    {
-        Ok(s) => s,
-        Err(e) => {
-            log::error!("KDB connection failed: {}", e);
-            return;
-        }
-    };
+    .await?;
 
     // Process incoming records
     while let Some((_time, batch)) = source.next().await {
@@ -121,11 +115,10 @@ async fn kdb_write_consumer<T>(
             ]);
 
             // Send sync message to ensure insert completes before continuing
-            if let Err(e) = socket.send_sync_message(&query).await {
-                log::error!("KDB insert failed: {}", e);
-            }
+            socket.send_sync_message(&query).await?;
         }
     }
+    Ok(())
 }
 
 /// Extension trait for writing streams to KDB+ tables.

wingfoil/src/nodes/async_io.rs

@@ -58,19 +58,19 @@ type ConsumerFunc<T, FUT> = Box<dyn FnOnce(Pin<Box<dyn FutStream<T>>>) -> FUT +
 pub(crate) struct AsyncConsumerNode<T, FUT>
 where
     T: Element + Send,
-    FUT: Future<Output = ()> + Send + 'static,
+    FUT: Future<Output = anyhow::Result<()>> + Send + 'static,
 {
     source: Rc<dyn Stream<T>>,
     sender: ChannelSender<T>,
     func: Option<ConsumerFunc<T, FUT>>,
-    handle: Option<tokio::task::JoinHandle<()>>,
+    handle: Option<tokio::task::JoinHandle<anyhow::Result<()>>>,
     rx: Option<ChannelReceiver<T>>,
 }
 
 impl<T, FUT> AsyncConsumerNode<T, FUT>
 where
     T: Element + Send,
-    FUT: Future<Output = ()> + Send + 'static,
+    FUT: Future<Output = anyhow::Result<()>> + Send + 'static,
 {
     pub fn new(source: Rc<dyn Stream<T>>, func: ConsumerFunc<T, FUT>) -> Self {
         let (sender, receiver) = channel_pair(None);
@@ -91,7 +91,7 @@ where
 impl<T, FUT> MutableNode for AsyncConsumerNode<T, FUT>
 where
     T: Element + Send,
-    FUT: Future<Output = ()> + Send + 'static,
+    FUT: Future<Output = anyhow::Result<()>> + Send + 'static,
 {
     fn cycle(&mut self, state: &mut GraphState) -> anyhow::Result<bool> {
         self.sender.send(state, self.source.peek_value())?;
@@ -119,7 +119,7 @@ where
                 .limit(run_mode, run_for)
                 .to_stream();
             let fut = func(Box::pin(src));
-            fut.await;
+            fut.await
         };
         let handle = state.tokio_runtime().spawn(f);
         self.handle = Some(handle);
@@ -133,7 +133,7 @@ where
 
     fn teardown(&mut self, state: &mut GraphState) -> anyhow::Result<()> {
         if let Some(handle) = self.handle.take() {
-            state.tokio_runtime().block_on(handle)?;
+            state.tokio_runtime().block_on(handle)??;
         }
         Ok(())
     }
@@ -405,6 +405,7 @@ mod tests {
                     while let Some((time, value)) = source.next().await {
                         println!("{time:?}, {value:?}");
                     }
+                    Ok(())
                 };
 
                 produce_async(example_producer)

wingfoil/src/nodes/mod.rs

@@ -250,7 +250,7 @@ pub trait StreamOperators<T: Element> {
     ) -> Rc<dyn Node>
     where
         T: Element + Send,
-        FUT: Future<Output = ()> + Send + 'static;
+        FUT: Future<Output = anyhow::Result<()>> + Send + 'static;
     fn finally<F: FnOnce(T, &GraphState) + 'static>(self: &Rc<Self>, func: F) -> Rc<dyn Node>;
     /// executes supplied closure on each tick
     fn for_each(self: &Rc<Self>, func: impl Fn(T, NanoTime) + 'static) -> Rc<dyn Node>;
@@ -410,7 +410,7 @@ where
     ) -> Rc<dyn Node>
     where
         T: Element + Send,
-        FUT: Future<Output = ()> + Send + 'static,
+        FUT: Future<Output = anyhow::Result<()>> + Send + 'static,
     {
         AsyncConsumerNode::new(self.clone(), func).into_node()
     }