AsyncProducer Handler update + version bump

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "messagebus"
-version = "0.9.10"
+version = "0.9.11"
 authors = ["Andrey Tkachenko <andrey@aidev.ru>"]
 repository = "https://github.com/andreytkachenko/messagebus.git"
 keywords = ["futures", "async", "tokio", "message", "bus"]
@@ -11,10 +11,7 @@ exclude = [".gitignore", ".cargo/config", ".github/**", "codecov.yml"]
 edition = "2018"
 
 [workspace]
-members = [
-    "crates/remote",
-    "crates/derive",
-]
+members = ["crates/remote", "crates/derive"]
 
 [dependencies]
 messagebus_derive = "0.2.5"
@@ -37,4 +34,10 @@ ctor = "0.1.21"
 anyhow = "1.0"
 env_logger = "0.9"
 serde_json = "1.0"
-tokio = { version = "1", features = ["macros", "parking_lot", "rt-multi-thread", "io-util", "sync"] }
+tokio = { version = "1", features = [
+  "macros",
+  "parking_lot",
+  "rt-multi-thread",
+  "io-util",
+  "sync",
+] }

src/handler.rs

@@ -1,20 +1,30 @@
 use core::iter::FromIterator;
-use std::ops::ControlFlow;
+use std::pin::Pin;
 
 use crate::{error::StdSyncSendError, Bus, Message};
 use async_trait::async_trait;
+use futures::Stream;
 
-#[async_trait]
-pub trait AsyncProducer<'a, M: Message>: Send {
-    type Error: StdSyncSendError;
-    type Context: Send + 'a;
-    type Response: Message;
-
-    async fn start(&'a self, msg: M, bus: &Bus) -> Result<Self::Context, Self::Error>;
-    async fn next(&'a self, ctx: &mut Self::Context, bus: &Bus) -> Result<ControlFlow<Self::Response>, Self::Error>;
-    async fn finish(&'a self, _ctx: Self::Context, _bus: &Bus) -> Result<(), Self::Error>;
+#[derive(Debug, Clone, Copy)]
+pub struct ProducerStats {
+    pub completed: usize,
+    pub failed: usize,
 }
 
+#[async_trait]
+pub trait AsyncProducer<M: Message>: Send + Sync {
+    type Item: Message;
+    type Response: Message;
+    type Error: StdSyncSendError;
+
+    async fn producer(
+        &self,
+        msg: M,
+        bus: &Bus,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<Self::Item, Self::Error>> + Send + '_>>, Self::Error>;
+
+    async fn finish(&self, stats: ProducerStats, bus: &Bus) -> Result<Self::Response, Self::Error>;
+}
 
 pub trait Handler<M: Message>: Send + Sync {
     type Error: StdSyncSendError;

src/receivers/synchronized/async.rs

@@ -97,7 +97,9 @@ where
     fn send(&self, mid: u64, m: M, req: bool, _bus: &Bus) -> Result<(), Error<M>> {
         match self.tx.send(Request::Request(mid, m, req)) {
             Ok(_) => Ok(()),
-            Err(mpsc::error::SendError(Request::Request(_, msg, _))) => Err(Error::send_closed(msg)),
+            Err(mpsc::error::SendError(Request::Request(_, msg, _))) => {
+                Err(Error::send_closed(msg))
+            }
             _ => unimplemented!(),
         }
     }