strev 0.6.0

Event-driven pub/sub messaging library with compile-time ack safety
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

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

use bytes::Bytes;
use strev::{HandlerResult, Message, Publisher, Router, ShutdownSignal, Topic};
use strev_channel::Channel;
use tokio_util::sync::CancellationToken;

#[tokio::test]
async fn router_processes_messages_end_to_end() {
    let channel = Channel::new(16);
    let topic_in = Topic::new("input");
    let count = Arc::new(AtomicU32::new(0));

    let mut router = Router::new();
    let count_clone = count.clone();

    router.add_handler(
        "test_handler",
        topic_in.clone(),
        channel.clone(),
        channel.clone(),
        move |msg: Message| {
            let c = count_clone.clone();
            async move {
                c.fetch_add(1, Ordering::SeqCst);
                Ok(HandlerResult::ack(msg))
            }
        },
    );

    let token = CancellationToken::new();
    let token_clone = token.clone();

    let router_handle =
        tokio::spawn(async move { router.run(ShutdownSignal::Token(token_clone)).await });

    tokio::time::sleep(Duration::from_millis(50)).await;

    let msg = Message::new(Bytes::from("test"));
    Publisher::publish(&channel, &topic_in, vec![msg])
        .await
        .unwrap();

    tokio::time::sleep(Duration::from_millis(100)).await;
    token.cancel();

    router_handle.await.unwrap().unwrap();
    assert_eq!(count.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn router_consumer_without_publisher() {
    let channel = Channel::new(16);
    let topic = Topic::new("events");
    let count = Arc::new(AtomicU32::new(0));

    let mut router = Router::new();
    let count_clone = count.clone();

    router.add_consumer(
        "consumer",
        topic.clone(),
        channel.clone(),
        move |msg: Message| {
            let c = count_clone.clone();
            async move {
                c.fetch_add(1, Ordering::SeqCst);
                Ok(HandlerResult::ack(msg))
            }
        },
    );

    let token = CancellationToken::new();
    let token_clone = token.clone();

    let router_handle =
        tokio::spawn(async move { router.run(ShutdownSignal::Token(token_clone)).await });

    tokio::time::sleep(Duration::from_millis(50)).await;

    for i in 0..5 {
        let msg = Message::new(Bytes::from(format!("msg-{i}")));
        Publisher::publish(&channel, &topic, vec![msg])
            .await
            .unwrap();
    }

    tokio::time::sleep(Duration::from_millis(200)).await;
    token.cancel();

    router_handle.await.unwrap().unwrap();
    assert_eq!(count.load(Ordering::SeqCst), 5);
}

#[tokio::test]
async fn router_shutdown_via_cancellation_token() {
    let channel = Channel::new(16);
    let topic = Topic::new("test");

    let mut router = Router::new();
    router.add_consumer("noop", topic, channel.clone(), |msg: Message| async move {
        Ok(HandlerResult::ack(msg))
    });

    let token = CancellationToken::new();
    let token_clone = token.clone();

    let handle = tokio::spawn(async move { router.run(ShutdownSignal::Token(token_clone)).await });

    tokio::time::sleep(Duration::from_millis(50)).await;
    token.cancel();

    let result = tokio::time::timeout(Duration::from_secs(2), handle).await;
    assert!(result.is_ok());
}