ramqp 0.6.6

Async AMQP 1.0 client for Rust on tokio — connects to RabbitMQ 4.x, ActiveMQ Artemis, and other AMQP 1.0 brokers.
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

//! Connection lifecycle events and their broadcast bus (WP-0.4).
//!
//! The [`EventBus`] lets applications observe connection health (including
//! reconnect transitions) without enabling `tracing` or any log backend.

use tokio::sync::broadcast;

/// A coarse, cloneable snapshot of connection health.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ConnectionState {
    /// The initial connect / handshake is in progress.
    Connecting,
    /// Open and healthy.
    Connected,
    /// Connection lost; the supervisor is attempting to re-establish.
    Reconnecting,
    /// Connected but degraded (e.g. operating from a bounded outbound buffer).
    Degraded,
    /// Permanently closed.
    Closed,
}

/// A connection lifecycle event published on the [`EventBus`].
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum ConnectionEvent {
    /// The initial connection / handshake started.
    Connecting,
    /// The connection opened successfully (or re-opened after a drop).
    Connected,
    /// A reconnect attempt is starting (1-based `attempt`).
    Reconnecting {
        /// The attempt number.
        attempt: u32,
    },
    /// The connection is degraded but still usable.
    Degraded {
        /// A short, human-readable reason.
        reason: String,
    },
    /// The connection closed.
    Closed {
        /// Whether the close was caused by an error (vs. a graceful close).
        error: bool,
        /// A short, human-readable reason.
        reason: String,
    },
}

impl ConnectionEvent {
    /// The coarse [`ConnectionState`] this event implies.
    pub fn state(&self) -> ConnectionState {
        match self {
            ConnectionEvent::Connecting => ConnectionState::Connecting,
            ConnectionEvent::Connected => ConnectionState::Connected,
            ConnectionEvent::Reconnecting { .. } => ConnectionState::Reconnecting,
            ConnectionEvent::Degraded { .. } => ConnectionState::Degraded,
            ConnectionEvent::Closed { .. } => ConnectionState::Closed,
        }
    }
}

/// A multi-producer/multi-consumer broadcast of [`ConnectionEvent`]s.
///
/// Subscribers that fall behind by more than the bus capacity observe a lag
/// notification (standard `broadcast` semantics) rather than blocking the
/// runtime — health events are lossy by design.
#[derive(Debug, Clone)]
pub struct EventBus {
    tx: broadcast::Sender<ConnectionEvent>,
}

impl EventBus {
    /// Create a bus buffering up to `capacity` recent events per subscriber.
    pub fn new(capacity: usize) -> Self {
        let (tx, _rx) = broadcast::channel(capacity);
        EventBus { tx }
    }

    /// Publish an event (ignored if there are no subscribers).
    pub fn publish(&self, event: ConnectionEvent) {
        let _ = self.tx.send(event);
    }

    /// Subscribe to subsequent events.
    pub fn subscribe(&self) -> broadcast::Receiver<ConnectionEvent> {
        self.tx.subscribe()
    }

    /// The current number of subscribers.
    pub fn subscriber_count(&self) -> usize {
        self.tx.receiver_count()
    }
}

impl Default for EventBus {
    fn default() -> Self {
        EventBus::new(64)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn subscribers_receive_lifecycle_sequence() {
        let bus = EventBus::new(16);
        let mut rx = bus.subscribe();
        bus.publish(ConnectionEvent::Connecting);
        bus.publish(ConnectionEvent::Connected);
        bus.publish(ConnectionEvent::Reconnecting { attempt: 1 });
        bus.publish(ConnectionEvent::Connected);

        assert_eq!(rx.recv().await.unwrap(), ConnectionEvent::Connecting);
        assert_eq!(rx.recv().await.unwrap(), ConnectionEvent::Connected);
        assert_eq!(
            rx.recv().await.unwrap().state(),
            ConnectionState::Reconnecting
        );
        assert_eq!(rx.recv().await.unwrap(), ConnectionEvent::Connected);
    }
}