rskit-messaging 0.2.0-alpha.3

Message broker abstractions, registry, middleware, and in-memory adapter
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

//! Test assertion and wait helpers over an [`InMemoryBroker`].

use std::time::Duration;

use super::broker::InMemoryBroker;
use crate::message::Message;

/// Assert that at least one message on `topic` satisfies the predicate.
///
/// # Panics
///
/// Panics when no matching message is found.
pub async fn assert_published<T: Clone + Send + Sync + 'static>(
    broker: &InMemoryBroker<T>,
    topic: &str,
    predicate: impl Fn(&Message<T>) -> bool,
) {
    let msgs = broker.messages(topic).await;
    assert!(
        msgs.iter().any(&predicate),
        "assert_published: no message on topic {topic:?} matched the predicate ({} checked)",
        msgs.len(),
    );
}

/// Assert that exactly `n` messages were published to `topic`.
///
/// # Panics
///
/// Panics when the count does not match.
pub async fn assert_published_n<T: Clone + Send + Sync + 'static>(
    broker: &InMemoryBroker<T>,
    topic: &str,
    n: usize,
) {
    let got = broker.message_count(topic).await;
    assert_eq!(
        got, n,
        "assert_published_n: topic {topic:?} has {got} messages, want {n}",
    );
}

/// Wait until at least one message appears on `topic` or the timeout expires.
///
/// Returns the first message on the topic.
///
/// # Panics
///
/// Panics if the timeout elapses before any message arrives.
pub async fn wait_for_message<T: Clone + Send + Sync + 'static>(
    broker: &InMemoryBroker<T>,
    topic: &str,
    timeout: Duration,
) -> Message<T> {
    let deadline = tokio::time::Instant::now() + timeout;

    loop {
        let msgs = broker.messages(topic).await;
        if let Some(m) = msgs.into_iter().next() {
            return m;
        }
        tokio::select! {
            () = broker.notified() => { /* re-check */ }
            () = tokio::time::sleep_until(deadline) => {
                panic!("wait_for_message: timed out after {timeout:?} waiting for message on topic {topic:?}");
            }
        }
    }
}

/// Assert that zero messages were published to `topic`.
///
/// # Panics
///
/// Panics when the topic is not empty.
pub async fn assert_no_messages<T: Clone + Send + Sync + 'static>(
    broker: &InMemoryBroker<T>,
    topic: &str,
) {
    let n = broker.message_count(topic).await;
    assert_eq!(
        n, 0,
        "assert_no_messages: topic {topic:?} has {n} messages, want 0",
    );
}