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

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

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

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

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

// Publishes Pong from inside the async handle method.
struct PingHandler;

impl EventHandler<Ping> for PingHandler {
    async fn handle(&self, _event: &Ping, bus: &EventBus) -> HandlerResult {
        bus.publish(Pong).await.map_err(|e| e.to_string())?;
        Ok(())
    }
}

struct PongHandler {
    count: Arc<AtomicUsize>,
    notify: Arc<tokio::sync::Notify>,
}

impl EventHandler<Pong> for PongHandler {
    async fn handle(&self, _event: &Pong, _bus: &EventBus) -> HandlerResult {
        self.count.fetch_add(1, Ordering::SeqCst);
        self.notify.notify_one();
        Ok(())
    }
}

#[tokio::test]
async fn async_handler_can_publish_to_bus() {
    let bus = EventBus::builder().build().await.expect("valid config");
    let pong_count = Arc::new(AtomicUsize::new(0));
    let notified = Arc::new(tokio::sync::Notify::new());

    let _ = bus.subscribe(PingHandler).await.expect("subscribe ping");
    let _ = bus
        .subscribe(PongHandler {
            count: Arc::clone(&pong_count),
            notify: Arc::clone(&notified),
        })
        .await
        .expect("subscribe pong");

    bus.publish(Ping).await.expect("publish ping");

    // Wait for PongHandler to fire before shutting down.
    tokio::time::timeout(std::time::Duration::from_secs(2), notified.notified())
        .await
        .expect("PongHandler did not fire within 2s");

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

    assert_eq!(pong_count.load(Ordering::SeqCst), 1, "pong handler should have fired once");
}

#[tokio::test]
async fn async_closure_handler_can_publish_to_bus() {
    let bus = EventBus::builder().build().await.expect("valid config");
    let pong_count = Arc::new(AtomicUsize::new(0));
    let notified = Arc::new(tokio::sync::Notify::new());

    let pong_count_clone = Arc::clone(&pong_count);
    let notify_clone = Arc::clone(&notified);
    let _ = bus
        .subscribe(move |_event: Pong, _bus: EventBus| {
            let count = Arc::clone(&pong_count_clone);
            let notify = Arc::clone(&notify_clone);
            async move {
                count.fetch_add(1, Ordering::SeqCst);
                notify.notify_one();
                Ok(())
            }
        })
        .await
        .expect("subscribe pong closure");

    let _ = bus
        .subscribe(|_event: Ping, bus: EventBus| async move {
            bus.publish(Pong).await.map_err(|e: jaeb::EventBusError| e.to_string())?;
            Ok(())
        })
        .await
        .expect("subscribe ping closure");

    bus.publish(Ping).await.expect("publish ping");

    // Wait for the pong closure handler to fire before shutting down.
    tokio::time::timeout(std::time::Duration::from_secs(2), notified.notified())
        .await
        .expect("pong closure handler did not fire within 2s");

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

    assert_eq!(pong_count.load(Ordering::SeqCst), 1, "pong closure handler should have fired once");
}

#[cfg(feature = "macros")]
mod macro_tests {
    use super::*;
    use jaeb::handler;

    #[handler]
    async fn on_ping(_event: &Ping, bus: &EventBus) -> HandlerResult {
        bus.publish(Pong).await.map_err(|e| e.to_string())?;
        Ok(())
    }

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

        let pong_count = Arc::new(AtomicUsize::new(0));
        let notified = Arc::new(tokio::sync::Notify::new());

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

        bus.publish(Ping).await.expect("publish ping");

        // Wait for PongHandler to fire before shutting down.
        tokio::time::timeout(std::time::Duration::from_secs(2), notified.notified())
            .await
            .expect("PongHandler did not fire within 2s");

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

        assert_eq!(
            pong_count.load(Ordering::SeqCst),
            1,
            "pong handler should have fired once via #[handler] macro"
        );
    }
}