monocoque_core/
monitor.rs

1//! Socket event monitoring.
2//!
3//! Provides event streams for tracking socket lifecycle events like
4//! connections, disconnections, and errors.
5
6use crate::endpoint::Endpoint;
7use std::fmt;
8
9/// Socket lifecycle events.
10#[derive(Debug, Clone)]
11pub enum SocketEvent {
12    /// Socket successfully connected to a peer.
13    Connected(Endpoint),
14
15    /// Socket disconnected from a peer.
16    Disconnected(Endpoint),
17
18    /// Socket successfully bound to an endpoint.
19    Bound(Endpoint),
20
21    /// Bind operation failed.
22    BindFailed { endpoint: Endpoint, reason: String },
23
24    /// Connection attempt failed.
25    ConnectFailed { endpoint: Endpoint, reason: String },
26
27    /// Socket is listening for incoming connections.
28    Listening(Endpoint),
29
30    /// Socket accepted a new incoming connection.
31    Accepted(Endpoint),
32}
33
34impl fmt::Display for SocketEvent {
35    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
36        match self {
37            Self::Connected(ep) => write!(f, "Connected to {ep}"),
38            Self::Disconnected(ep) => write!(f, "Disconnected from {ep}"),
39            Self::Bound(ep) => write!(f, "Bound to {ep}"),
40            Self::BindFailed { endpoint, reason } => {
41                write!(f, "Bind failed for {endpoint}: {reason}")
42            }
43            Self::ConnectFailed { endpoint, reason } => {
44                write!(f, "Connect failed for {endpoint}: {reason}")
45            }
46            Self::Listening(ep) => write!(f, "Listening on {ep}"),
47            Self::Accepted(ep) => write!(f, "Accepted connection from {ep}"),
48        }
49    }
50}
51
52/// Handle for receiving socket events.
53///
54/// This is a channel receiver that provides a stream of socket lifecycle events.
55pub type SocketMonitor = flume::Receiver<SocketEvent>;
56
57/// Internal sender for socket events.
58///
59/// This is exposed publicly to allow socket implementations to emit events.
60pub type SocketEventSender = flume::Sender<SocketEvent>;
61
62/// Bound on the number of undrained monitor events held in the channel.
63///
64/// Lifecycle events (connect/disconnect/bind/accept) are low volume, but the
65/// channel was previously unbounded, so an application that never drained its
66/// monitor would let it grow without limit. Bounding it caps that footprint;
67/// [`emit`] drops events (rather than blocking the socket path) once full.
68pub const MONITOR_CHANNEL_CAP: usize = 256;
69
70/// Creates a new monitoring channel pair.
71///
72/// The channel is bounded by [`MONITOR_CHANNEL_CAP`]. Emit events with [`emit`],
73/// which never blocks the caller.
74///
75/// This is exposed publicly to allow socket implementations to create monitors.
76#[must_use]
77pub fn create_monitor() -> (SocketEventSender, SocketMonitor) {
78    flume::bounded(MONITOR_CHANNEL_CAP)
79}
80
81/// Emit a monitor event without ever blocking the socket path.
82///
83/// Uses a non-blocking send: if the monitor is full (the application is not
84/// draining it) or the receiver has been dropped, the event is discarded rather
85/// than stalling the socket operation that produced it.
86pub fn emit(sender: &SocketEventSender, event: SocketEvent) {
87    let _ = sender.try_send(event);
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93    use std::net::SocketAddr;
94
95    #[test]
96    fn test_socket_event_display() {
97        let addr: SocketAddr = "127.0.0.1:5555".parse().unwrap();
98        let event = SocketEvent::Connected(Endpoint::Tcp(addr));
99        assert_eq!(event.to_string(), "Connected to tcp://127.0.0.1:5555");
100    }
101
102    #[test]
103    fn test_monitor_channel() {
104        let (sender, receiver) = create_monitor();
105        let addr: SocketAddr = "127.0.0.1:5555".parse().unwrap();
106        emit(&sender, SocketEvent::Connected(Endpoint::Tcp(addr)));
107
108        let event = receiver.recv().unwrap();
109        assert!(matches!(event, SocketEvent::Connected(_)));
110    }
111
112    #[test]
113    fn emit_is_bounded_and_never_blocks_when_undrained() {
114        // Fill well past the cap without ever draining the receiver. emit must
115        // not block or grow the channel beyond MONITOR_CHANNEL_CAP.
116        let (sender, receiver) = create_monitor();
117        let addr: SocketAddr = "127.0.0.1:5555".parse().unwrap();
118        for _ in 0..(MONITOR_CHANNEL_CAP * 4) {
119            emit(&sender, SocketEvent::Connected(Endpoint::Tcp(addr)));
120        }
121        assert_eq!(
122            receiver.len(),
123            MONITOR_CHANNEL_CAP,
124            "monitor channel must stay bounded at its cap when undrained"
125        );
126    }
127}
monocoque_core/monitor.rs

monocoque_core/
monitor.rs
