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

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

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

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

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

struct Counter {
    count: Arc<AtomicUsize>,
}

impl SyncEventHandler<Tick> for Counter {
    fn handle(&self, _event: &Tick) -> HandlerResult {
        self.count.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
}

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

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

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

    bus.publish(Tick).await.expect("publish");
    assert_eq!(count.load(Ordering::SeqCst), 1);

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

    bus.publish(Tick).await.expect("publish after unsubscribe");
    assert_eq!(count.load(Ordering::SeqCst), 1); // unchanged

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

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

    let sub = bus.subscribe(Counter { count: Arc::clone(&count) }).await.expect("subscribe");
    let id = sub.id();

    bus.publish(Tick).await.expect("publish");
    assert_eq!(count.load(Ordering::SeqCst), 1);

    // Use the bus-level unsubscribe with the raw id.
    let removed = bus.unsubscribe(id).await.expect("unsubscribe by id");
    assert!(removed);

    bus.publish(Tick).await.expect("publish after unsubscribe");
    assert_eq!(count.load(Ordering::SeqCst), 1);

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

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

    // Subscribe and immediately unsubscribe to obtain a valid-but-removed id.
    let sub = bus
        .subscribe(Counter {
            count: Arc::new(AtomicUsize::new(0)),
        })
        .await
        .expect("subscribe");
    let id = sub.id();
    sub.unsubscribe().await.expect("first unsubscribe");

    // Now `id` is no longer tracked by the runtime state — should return Ok(false).
    let removed = bus.unsubscribe(id).await.expect("second unsubscribe");
    assert!(!removed);

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

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

    let sub = bus.subscribe(Counter { count: Arc::clone(&count) }).await.expect("subscribe");
    let id = sub.id();

    let first = sub.unsubscribe().await.expect("first unsubscribe");
    assert!(first);

    let second = bus.unsubscribe(id).await.expect("second unsubscribe");
    assert!(!second);

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

/// Concurrent publish and unsubscribe should not panic or deadlock.
///
/// This test exercises the race between publish (which snapshots the
/// listener list) and unsubscribe (which mutates it). The bus uses
/// copy-on-write (`Arc<Vec<Listener>>`) so both operations are safe.
#[tokio::test]
async fn concurrent_publish_and_unsubscribe_is_safe() {
    let bus = EventBus::builder().buffer_size(256).build().await.expect("valid config");
    let count = Arc::new(AtomicUsize::new(0));

    // Register several listeners.
    let mut subs = Vec::new();
    for _ in 0..10 {
        let sub = bus.subscribe(Counter { count: Arc::clone(&count) }).await.expect("subscribe");
        subs.push(sub);
    }

    let mut tasks = tokio::task::JoinSet::new();

    // Spawn concurrent publishers.
    let bus_pub = bus.clone();
    tasks.spawn(async move {
        for _ in 0..100 {
            let _ = bus_pub.publish(Tick).await;
            tokio::task::yield_now().await;
        }
    });

    // Concurrently unsubscribe half the listeners.
    let bus_unsub = bus.clone();
    tasks.spawn(async move {
        for sub in subs.drain(..5) {
            let _ = bus_unsub.unsubscribe(sub.id()).await;
            tokio::task::yield_now().await;
        }
    });

    // All tasks should complete promptly and without panic.
    tokio::time::timeout(Duration::from_secs(3), async {
        while let Some(result) = tasks.join_next().await {
            result.expect("concurrent task should not panic");
        }
    })
    .await
    .expect("concurrent publish/unsubscribe timed out");

    // Bus remains operational after the race.
    bus.publish(Tick).await.expect("publish after race should succeed");
    assert!(bus.is_healthy().await, "bus should remain healthy after race");

    // The counter should have been incremented some number of times — we
    // don't assert an exact count because the ordering is non-deterministic.
    assert!(count.load(Ordering::SeqCst) > 0, "at least some events should have been delivered");

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