init

event_bus/src/bus.rs

@@ -331,6 +331,48 @@ impl EventBus {
         })
     }
 
+    // todo : Undocumented
+    pub fn on_async_with<T, C, F, Fut>(&self, topic: &str, context: C, handler: F) -> JoinHandle<()>
+    where
+        T: Any + Send + Sync + Clone + 'static,
+        C: Clone + Send + Sync + 'static,
+        F: Fn(C, T) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = ()> + Send + 'static,
+    {
+        let mut rx = self.get_or_create_sender(topic).subscribe();
+        let topic_owned = topic.to_string();
+
+        debug!(topic, "Async subscriber registered");
+
+        tokio::spawn(async move {
+            loop {
+                match rx.recv().await {
+                    Ok(evt) => {
+                        if let Some(typed) = evt.downcast_ref::<T>() {
+                            trace!(topic = topic_owned, "Async handler invoked");
+
+                            handler(context.clone(), typed.clone()).await;
+                        }
+                    }
+                    Err(broadcast::error::RecvError::Lagged(n)) => {
+                        warn!(
+                            topic = topic_owned,
+                            skipped = n,
+                            "Subscriber lagged, messages dropped"
+                        );
+                    }
+                    Err(broadcast::error::RecvError::Closed) => {
+                        debug!(
+                            topic = topic_owned,
+                            "Channel closed, async subscriber exiting"
+                        );
+                        break;
+                    }
+                }
+            }
+        })
+    }
+
     // ─────────────────────────────────────────────────────────────────────────
     // Subscription — low-level access (advanced use cases)
     // ─────────────────────────────────────────────────────────────────────────