oxicode_sdk/ports/inmem/
event.rs

1//! In-process `EventBus` impls.
2//!
3//! - `InProcessEventBus`: tokio broadcast channel (no broker, no network)
4//! - `NullEventBus`: accepts publish, drops everything (for tests)
5
6use std::future::Future;
7use std::pin::Pin;
8use std::sync::Arc;
9
10use crate::SdkError;
11use crate::ports::{EventBus, EventPayload, EventTopic, SubscriptionHandle};
12
13/// In-process event bus using `tokio::sync::broadcast`.
14///
15/// Capacity is the broadcast buffer. Slow consumers may lag.
16pub struct InProcessEventBus {
17    tx: tokio::sync::broadcast::Sender<(EventTopic, EventPayload)>,
18}
19
20impl std::fmt::Debug for InProcessEventBus {
21    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
22        f.debug_struct("InProcessEventBus").finish()
23    }
24}
25
26impl InProcessEventBus {
27    /// Create a new bus with the given broadcast capacity.
28    pub fn new(capacity: usize) -> Arc<Self> {
29        let (tx, _) = tokio::sync::broadcast::channel(capacity);
30        Arc::new(Self { tx })
31    }
32}
33
34impl EventBus for InProcessEventBus {
35    fn publish(
36        &self,
37        topic: &EventTopic,
38        payload: EventPayload,
39    ) -> Pin<Box<dyn Future<Output = Result<(), SdkError>> + Send + '_>> {
40        // Best-effort: no active subscribers is not an error.
41        let _ = self.tx.send((topic.clone(), payload));
42        Box::pin(async { Ok(()) })
43    }
44
45    fn subscribe(
46        &self,
47        _topic: &EventTopic,
48    ) -> Pin<Box<dyn Future<Output = Result<SubscriptionHandle, SdkError>> + Send + '_>> {
49        let mut rx = self.tx.subscribe();
50        let (tx, rx2) = tokio::sync::mpsc::channel(64);
51        tokio::spawn(async move {
52            while let Ok(event) = rx.recv().await {
53                if tx.send(event).await.is_err() {
54                    break;
55                }
56            }
57        });
58        Box::pin(async { Ok(SubscriptionHandle::from_receiver(rx2)) })
59    }
60}
61
62#[cfg(test)]
63mod tests {
64    use super::*;
65    use serde_json::json;
66
67    #[tokio::test]
68    async fn publish_then_receive() {
69        let bus = InProcessEventBus::new(8);
70        bus.publish(&"kernel.agent.started".to_string(), json!({"id": "a1"}))
71            .await
72            .unwrap();
73        let mut sub = bus.subscribe(&"kernel".to_string()).await.unwrap();
74        bus.publish(&"kernel.tool.completed".to_string(), json!({"k": 1}))
75            .await
76            .unwrap();
77        let (topic, payload) = sub.recv().await.unwrap();
78        assert_eq!(topic, "kernel.tool.completed");
79        assert_eq!(payload, json!({"k": 1}));
80    }
81
82    #[tokio::test]
83    async fn multiple_subscribers() {
84        let bus = InProcessEventBus::new(8);
85        let mut s1 = bus.subscribe(&"x".to_string()).await.unwrap();
86        let mut s2 = bus.subscribe(&"x".to_string()).await.unwrap();
87        bus.publish(&"x".to_string(), json!(1)).await.unwrap();
88        let (_, p1) = s1.recv().await.unwrap();
89        let (_, p2) = s2.recv().await.unwrap();
90        assert_eq!(p1, json!(1));
91        assert_eq!(p2, json!(1));
92    }
93}
oxicode_sdk/ports/inmem/event.rs

oxicode_sdk/ports/inmem/
event.rs
