soaprs_memory/
events.rs

1//! In-process typed event delivery.
2
3use std::{
4    fmt,
5    sync::{
6        Arc, RwLock,
7        atomic::{AtomicU64, Ordering},
8    },
9};
10
11use soaprs_core::{BoxFuture, SoapError, SoapResult};
12use soaprs_events::{Event, EventEnvelope, EventHandler, EventPublisher};
13
14/// Opaque identity of an in-memory event subscription.
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
16pub struct SubscriptionId(u64);
17
18impl SubscriptionId {
19    /// Returns the numeric reference-adapter identity.
20    pub const fn get(self) -> u64 {
21        self.0
22    }
23}
24
25struct Subscription<E>
26where
27    E: Event,
28{
29    id: SubscriptionId,
30    handler: Arc<dyn EventHandler<E>>,
31}
32
33/// Subscription-order event bus for tests and single-process applications.
34///
35/// Publishing stops and returns the first handler error. It never silently
36/// logs and discards a failed handler invocation.
37pub struct MemoryEventBus<E>
38where
39    E: Event,
40{
41    next_id: AtomicU64,
42    subscriptions: RwLock<Vec<Subscription<E>>>,
43}
44
45impl<E> MemoryEventBus<E>
46where
47    E: Event,
48{
49    /// Creates an empty event bus.
50    pub const fn new() -> Self {
51        Self {
52            next_id: AtomicU64::new(1),
53            subscriptions: RwLock::new(Vec::new()),
54        }
55    }
56
57    /// Adds a handler.
58    pub fn subscribe(&self, handler: Arc<dyn EventHandler<E>>) -> SoapResult<SubscriptionId> {
59        let id = SubscriptionId(self.next_id.fetch_add(1, Ordering::Relaxed));
60        self.subscriptions
61            .write()
62            .map_err(|_| SoapError::infrastructure("in-memory event bus write lock poisoned"))?
63            .push(Subscription { id, handler });
64        Ok(id)
65    }
66
67    /// Removes a handler and reports whether it existed.
68    pub fn unsubscribe(&self, id: SubscriptionId) -> SoapResult<bool> {
69        let mut subscriptions = self
70            .subscriptions
71            .write()
72            .map_err(|_| SoapError::infrastructure("in-memory event bus write lock poisoned"))?;
73        let Some(index) = subscriptions.iter().position(|entry| entry.id == id) else {
74            return Ok(false);
75        };
76        subscriptions.remove(index);
77        Ok(true)
78    }
79
80    /// Returns the number of current handlers.
81    pub fn subscriber_count(&self) -> SoapResult<usize> {
82        Ok(self
83            .subscriptions
84            .read()
85            .map_err(|_| SoapError::infrastructure("in-memory event bus read lock poisoned"))?
86            .len())
87    }
88}
89
90impl<E> Default for MemoryEventBus<E>
91where
92    E: Event,
93{
94    fn default() -> Self {
95        Self::new()
96    }
97}
98
99impl<E> fmt::Debug for MemoryEventBus<E>
100where
101    E: Event,
102{
103    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
104        formatter
105            .debug_struct("MemoryEventBus")
106            .field("event_family", &std::any::type_name::<E>())
107            .finish_non_exhaustive()
108    }
109}
110
111impl<E> EventPublisher<E> for MemoryEventBus<E>
112where
113    E: Event,
114{
115    fn publish(&self, event: EventEnvelope<E>) -> BoxFuture<'_, SoapResult<()>> {
116        let handlers = match self.subscriptions.read() {
117            Ok(subscriptions) => subscriptions
118                .iter()
119                .map(|entry| Arc::clone(&entry.handler))
120                .collect::<Vec<_>>(),
121            Err(_) => {
122                return Box::pin(async {
123                    Err(SoapError::infrastructure(
124                        "in-memory event bus read lock poisoned",
125                    ))
126                });
127            }
128        };
129
130        Box::pin(async move {
131            for handler in handlers {
132                handler.handle(&event).await?;
133            }
134            Ok(())
135        })
136    }
137}
138
139#[cfg(test)]
140mod tests {
141    use std::{
142        sync::{Arc, Mutex},
143        time::UNIX_EPOCH,
144    };
145
146    use soaprs_contract_tests::block_on;
147    use soaprs_core::{BoxFuture, MessageMetadata, SoapError, SoapResult};
148    use soaprs_events::{DomainEvent, Event, EventEnvelope, EventHandler, EventPublisher};
149
150    use super::MemoryEventBus;
151
152    #[derive(Debug)]
153    struct TestEvent(u8);
154
155    impl Event for TestEvent {
156        fn event_type(&self) -> &'static str {
157            "contract.bus-event"
158        }
159    }
160
161    impl DomainEvent for TestEvent {}
162
163    struct RecordingHandler {
164        seen: Arc<Mutex<Vec<u8>>>,
165        fails: bool,
166    }
167
168    impl EventHandler<TestEvent> for RecordingHandler {
169        fn handle<'a>(
170            &'a self,
171            event: &'a EventEnvelope<TestEvent>,
172        ) -> BoxFuture<'a, SoapResult<()>> {
173            Box::pin(async move {
174                self.seen
175                    .lock()
176                    .map_err(|_| SoapError::infrastructure("test event lock poisoned"))?
177                    .push(event.message.0);
178                if self.fails {
179                    Err(SoapError::domain("event handler failed"))
180                } else {
181                    Ok(())
182                }
183            })
184        }
185    }
186
187    #[test]
188    fn bus_preserves_order_and_surfaces_handler_failure() {
189        let bus = MemoryEventBus::new();
190        let seen = Arc::new(Mutex::new(Vec::new()));
191        for fails in [false, true, false] {
192            let subscribed = bus.subscribe(Arc::new(RecordingHandler {
193                seen: Arc::clone(&seen),
194                fails,
195            }));
196            assert!(subscribed.is_ok(), "{subscribed:?}");
197        }
198
199        let result = block_on(bus.publish(EventEnvelope::new(
200            TestEvent(7),
201            MessageMetadata::new("event-1", UNIX_EPOCH),
202        )));
203        assert!(result.is_err());
204        assert_eq!(
205            seen.lock().ok().map(|values| values.clone()),
206            Some(vec![7, 7])
207        );
208    }
209}
soaprs_memory/events.rs

soaprs_memory/
events.rs
