jaeb 0.3.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

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

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

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

#[derive(Clone, Debug)]
struct Ping {
    value: usize,
}

#[derive(Clone, Debug)]
struct Pong {
    value: usize,
}

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

struct AsyncCounter {
    count: Arc<AtomicUsize>,
}

impl EventHandler<Ping> for AsyncCounter {
    async fn handle(&self, event: &Ping) -> HandlerResult {
        self.count.fetch_add(event.value, Ordering::SeqCst);
        Ok(())
    }
}

struct SyncCounter {
    count: Arc<AtomicUsize>,
}

impl SyncEventHandler<Ping> for SyncCounter {
    fn handle(&self, event: &Ping) -> HandlerResult {
        self.count.fetch_add(event.value, Ordering::SeqCst);
        Ok(())
    }
}

struct PongCounter {
    count: Arc<AtomicUsize>,
}

impl SyncEventHandler<Pong> for PongCounter {
    fn handle(&self, event: &Pong) -> HandlerResult {
        self.count.fetch_add(event.value, Ordering::SeqCst);
        Ok(())
    }
}

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

#[tokio::test]
async fn async_handler_receives_event() {
    let bus = EventBus::new(16).expect("valid config");
    let count = Arc::new(AtomicUsize::new(0));

    let _ = bus.subscribe(AsyncCounter { count: Arc::clone(&count) }).await.expect("subscribe");

    bus.publish(Ping { value: 42 }).await.expect("publish");

    // Shutdown drains in-flight async handlers deterministically.
    bus.shutdown().await.expect("shutdown");

    assert_eq!(count.load(Ordering::SeqCst), 42);
}

#[tokio::test]
async fn sync_handler_receives_event() {
    let bus = EventBus::new(16).expect("valid config");
    let count = Arc::new(AtomicUsize::new(0));

    let _ = bus.subscribe(SyncCounter { count: Arc::clone(&count) }).await.expect("subscribe");

    bus.publish(Ping { value: 7 }).await.expect("publish");

    // Sync handlers complete before publish returns.
    assert_eq!(count.load(Ordering::SeqCst), 7);
    bus.shutdown().await.expect("shutdown");
}

#[tokio::test]
async fn publish_with_no_listeners_is_noop() {
    let bus = EventBus::new(16).expect("valid config");

    // Publishing with no registered handlers should not error or panic.
    bus.publish(Ping { value: 99 }).await.expect("publish with no listeners");

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

#[tokio::test]
async fn multiple_handlers_same_event_all_receive() {
    let bus = EventBus::new(16).expect("valid config");
    let sync_count = Arc::new(AtomicUsize::new(0));
    let async_count_a = Arc::new(AtomicUsize::new(0));
    let async_count_b = Arc::new(AtomicUsize::new(0));

    let _ = bus
        .subscribe(SyncCounter {
            count: Arc::clone(&sync_count),
        })
        .await
        .expect("subscribe sync");

    let _ = bus
        .subscribe(AsyncCounter {
            count: Arc::clone(&async_count_a),
        })
        .await
        .expect("subscribe async a");

    let _ = bus
        .subscribe(AsyncCounter {
            count: Arc::clone(&async_count_b),
        })
        .await
        .expect("subscribe async b");

    bus.publish(Ping { value: 5 }).await.expect("publish");

    // Shutdown drains in-flight async handlers deterministically.
    bus.shutdown().await.expect("shutdown");

    assert_eq!(sync_count.load(Ordering::SeqCst), 5);
    assert_eq!(async_count_a.load(Ordering::SeqCst), 5);
    assert_eq!(async_count_b.load(Ordering::SeqCst), 5);
}

#[tokio::test]
async fn handlers_only_receive_their_event_type() {
    let bus = EventBus::new(16).expect("valid config");
    let ping_count = Arc::new(AtomicUsize::new(0));
    let pong_count = Arc::new(AtomicUsize::new(0));

    let _ = bus
        .subscribe(SyncCounter {
            count: Arc::clone(&ping_count),
        })
        .await
        .expect("subscribe ping");

    let _ = bus
        .subscribe(PongCounter {
            count: Arc::clone(&pong_count),
        })
        .await
        .expect("subscribe pong");

    // Only publish Ping — Pong handler must not fire.
    bus.publish(Ping { value: 10 }).await.expect("publish ping");

    assert_eq!(ping_count.load(Ordering::SeqCst), 10);
    assert_eq!(pong_count.load(Ordering::SeqCst), 0);

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