jaeb 0.3.1

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

use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;

use jaeb::{DeadLetter, EventBus, EventHandler, HandlerResult, SyncEventHandler};
use tokio::sync::Notify;

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

#[derive(Clone, Debug)]
struct Task(#[allow(dead_code)] String);

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

struct NamedSyncHandler;

impl SyncEventHandler<Task> for NamedSyncHandler {
    fn handle(&self, _event: &Task) -> HandlerResult {
        Err("named sync handler fails".into())
    }

    fn name(&self) -> Option<&'static str> {
        Some("audit-logger")
    }
}

struct NamedAsyncHandler;

impl EventHandler<Task> for NamedAsyncHandler {
    async fn handle(&self, _event: &Task) -> HandlerResult {
        Err("named async handler fails".into())
    }

    fn name(&self) -> Option<&'static str> {
        Some("async-worker")
    }
}

struct UnnamedSyncHandler;

impl SyncEventHandler<Task> for UnnamedSyncHandler {
    fn handle(&self, _event: &Task) -> HandlerResult {
        Err("unnamed handler fails".into())
    }
}

struct DeadLetterCollector {
    letters: Arc<Mutex<Vec<DeadLetter>>>,
    notify: Arc<Notify>,
}

impl SyncEventHandler<DeadLetter> for DeadLetterCollector {
    fn handle(&self, event: &DeadLetter) -> HandlerResult {
        let mut guard = self.letters.lock().unwrap();
        guard.push(event.clone());
        self.notify.notify_one();
        Ok(())
    }
}

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

#[tokio::test]
async fn named_sync_handler_appears_in_dead_letter() {
    let bus = EventBus::new(16).expect("valid config");
    let notify = Arc::new(Notify::new());
    let letters: Arc<Mutex<Vec<DeadLetter>>> = Arc::default();

    let _ = bus
        .subscribe_dead_letters(DeadLetterCollector {
            letters: Arc::clone(&letters),
            notify: Arc::clone(&notify),
        })
        .await
        .expect("subscribe dead letters");

    let _ = bus.subscribe(NamedSyncHandler).await.expect("subscribe");

    bus.publish(Task("test".into())).await.expect("publish");

    tokio::time::timeout(Duration::from_secs(1), notify.notified())
        .await
        .expect("timed out waiting for dead letter");

    let guard = letters.lock().unwrap();
    assert_eq!(guard.len(), 1);
    assert_eq!(guard[0].listener_name, Some("audit-logger"));
}

#[tokio::test]
async fn named_async_handler_appears_in_dead_letter() {
    let bus = EventBus::new(16).expect("valid config");
    let letters: Arc<Mutex<Vec<DeadLetter>>> = Arc::default();

    let _ = bus
        .subscribe_dead_letters(DeadLetterCollector {
            letters: Arc::clone(&letters),
            notify: Arc::new(Notify::new()),
        })
        .await
        .expect("subscribe dead letters");

    let _ = bus.subscribe(NamedAsyncHandler).await.expect("subscribe");

    bus.publish(Task("test".into())).await.expect("publish");

    // Shutdown drains in-flight async tasks and delivers dead letters directly.
    bus.shutdown().await.expect("shutdown");

    let guard = letters.lock().unwrap();
    assert_eq!(guard.len(), 1);
    assert_eq!(guard[0].listener_name, Some("async-worker"));
}

#[tokio::test]
async fn unnamed_handler_defaults_to_none() {
    let bus = EventBus::new(16).expect("valid config");
    let notify = Arc::new(Notify::new());
    let letters: Arc<Mutex<Vec<DeadLetter>>> = Arc::default();

    let _ = bus
        .subscribe_dead_letters(DeadLetterCollector {
            letters: Arc::clone(&letters),
            notify: Arc::clone(&notify),
        })
        .await
        .expect("subscribe dead letters");

    let _ = bus.subscribe(UnnamedSyncHandler).await.expect("subscribe");

    bus.publish(Task("test".into())).await.expect("publish");

    tokio::time::timeout(Duration::from_secs(1), notify.notified())
        .await
        .expect("timed out waiting for dead letter");

    let guard = letters.lock().unwrap();
    assert_eq!(guard.len(), 1);
    assert_eq!(guard[0].listener_name, None);
}