jaeb 0.4.0

simple snapshot-driven event bus
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use jaeb::{EventBus, EventHandler, HandlerResult, SyncEventHandler};

// ── Event types ──────────────────────────────────────────────────────

#[derive(Clone, Debug)]
struct EventA;

#[derive(Clone, Debug)]
struct EventB;

// ── Handlers ─────────────────────────────────────────────────────────

struct SyncHandlerA;
impl SyncEventHandler<EventA> for SyncHandlerA {
    fn handle(&self, _event: &EventA) -> HandlerResult {
        Ok(())
    }
    fn name(&self) -> Option<&'static str> {
        Some("handler-a")
    }
}

struct SyncHandlerB;
impl SyncEventHandler<EventB> for SyncHandlerB {
    fn handle(&self, _event: &EventB) -> HandlerResult {
        Ok(())
    }
}

struct SlowAsyncHandler {
    started: Arc<AtomicBool>,
}
impl EventHandler<EventA> for SlowAsyncHandler {
    async fn handle(&self, _event: &EventA) -> HandlerResult {
        self.started.store(true, Ordering::SeqCst);
        tokio::time::sleep(Duration::from_secs(10)).await;
        Ok(())
    }
    fn name(&self) -> Option<&'static str> {
        Some("slow-async")
    }
}

// ── Tests ────────────────────────────────────────────────────────────

#[tokio::test]
async fn stats_empty_bus() {
    let bus = EventBus::builder().buffer_size(128).build().await.expect("valid config");

    let stats = bus.stats().await.expect("stats");
    assert_eq!(stats.total_subscriptions, 0);
    assert!(stats.registered_event_types.is_empty());
    assert!(stats.subscriptions_by_event.is_empty());
    assert_eq!(stats.queue_capacity, 128);
    assert_eq!(stats.publish_permits_available, 128);
    assert_eq!(stats.publish_in_flight, 0);
    assert_eq!(stats.in_flight_async, 0);
    assert!(!stats.shutdown_called);

    bus.shutdown().await.expect("shutdown");
}

#[tokio::test]
async fn stats_after_subscriptions() {
    let bus = EventBus::builder().buffer_size(64).build().await.expect("valid config");

    let _sub_a = bus.subscribe(SyncHandlerA).await.expect("subscribe a");
    let _sub_b = bus.subscribe(SyncHandlerB).await.expect("subscribe b");

    let stats = bus.stats().await.expect("stats");
    assert_eq!(stats.total_subscriptions, 2);
    assert_eq!(stats.registered_event_types.len(), 2);

    // EventA should have 1 listener named "handler-a"
    let event_a_name = stats
        .registered_event_types
        .iter()
        .find(|n| n.contains("EventA"))
        .expect("EventA should be registered");
    let a_listeners = &stats.subscriptions_by_event[event_a_name];
    assert_eq!(a_listeners.len(), 1);
    assert_eq!(a_listeners[0].name, Some("handler-a"));

    // EventB should have 1 unnamed listener
    let event_b_name = stats
        .registered_event_types
        .iter()
        .find(|n| n.contains("EventB"))
        .expect("EventB should be registered");
    let b_listeners = &stats.subscriptions_by_event[event_b_name];
    assert_eq!(b_listeners.len(), 1);
    assert_eq!(b_listeners[0].name, None);

    bus.shutdown().await.expect("shutdown");
}

#[tokio::test]
async fn stats_in_flight_async() {
    let bus = EventBus::builder()
        .buffer_size(64)
        .shutdown_timeout(Duration::from_millis(100))
        .build()
        .await
        .expect("valid config");
    let started = Arc::new(AtomicBool::new(false));

    let _sub = bus
        .subscribe(SlowAsyncHandler {
            started: Arc::clone(&started),
        })
        .await
        .expect("subscribe");

    bus.publish(EventA).await.expect("publish");

    // Wait for the async handler to start running.
    tokio::time::timeout(Duration::from_secs(2), async {
        while !started.load(Ordering::SeqCst) {
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    })
    .await
    .expect("handler should have started");

    let stats = bus.stats().await.expect("stats");
    assert!(
        stats.in_flight_async >= 1,
        "expected at least 1 in-flight async task, got {}",
        stats.in_flight_async
    );

    // Shutdown with timeout — the slow handler will be aborted.
    let _ = bus.shutdown().await;
}

#[tokio::test]
async fn stats_shutdown_called() {
    let bus = EventBus::builder().buffer_size(64).build().await.expect("valid config");

    let stats = bus.stats().await.expect("stats before shutdown");
    assert!(!stats.shutdown_called);

    bus.shutdown().await.expect("shutdown");

    // After shutdown, stats() should return Stopped.
    let result = bus.stats().await;
    assert!(result.is_err(), "stats after shutdown should fail");
}

#[tokio::test]
async fn stats_after_unsubscribe() {
    let bus = EventBus::builder().buffer_size(64).build().await.expect("valid config");

    let sub = bus.subscribe(SyncHandlerA).await.expect("subscribe");

    let stats = bus.stats().await.expect("stats");
    assert_eq!(stats.total_subscriptions, 1);

    sub.unsubscribe().await.expect("unsubscribe");

    let stats = bus.stats().await.expect("stats");
    assert_eq!(stats.total_subscriptions, 0);
    // Event type should still be tracked but with empty listener list removed.
    // After removing the last listener for a type, the entry is cleaned up.
    assert!(stats.registered_event_types.is_empty());

    bus.shutdown().await.expect("shutdown");
}