rzmq 0.5.23

High performance, CPU and memory efficient, fully asynchronous, safe pure-Rust implementation of ZeroMQ (ØMQ) messaging with io_uring and TCP Cork acceleration on Linux.
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

use std::time::Duration;

/// Represents significant events occurring within a socket or its connections.
/// Inspired by libzmq socket monitor events.
#[derive(Debug, Clone)]
#[non_exhaustive] // Allow adding more events later
pub enum SocketEvent {
  // --- Listener Events ---
  /// Socket has started listening successfully on the endpoint.
  Listening { endpoint: String },
  /// Socket failed to bind to the endpoint.
  BindFailed { endpoint: String, error_msg: String },
  /// Accepted a new connection. `peer_addr` is the address of the remote peer.
  Accepted { endpoint: String, peer_addr: String },
  /// Failed to accept a new connection.
  AcceptFailed { endpoint: String, error_msg: String },

  // --- Connecter Events ---
  /// Connection attempt established (transport layer). Handshake may follow.
  /// Note: Often handshake success is the more useful event.
  Connected { endpoint: String, peer_addr: String },
  /// Initial connection attempt failed, retrying starts (if configured).
  ConnectDelayed { endpoint: String, error_msg: String },
  /// Retrying connection after delay.
  ConnectRetried {
    endpoint: String,
    interval: Duration,
  },
  /// Connection attempt failed definitively after retries or immediately.
  ConnectFailed { endpoint: String, error_msg: String },

  // --- General Connection/Session Events ---
  /// Connection closed (listener stopped, connection dropped).
  Closed { endpoint: String },
  /// Failed to close a connection cleanly (should be rare).
  // CloseFailed { endpoint: String, error: ZmqError },
  /// Peer disconnected or connection terminated. Endpoint identifies the peer connection URI.
  Disconnected { endpoint: String },

  // --- Handshake/Security Events (Often more useful than raw Connected/Accepted) ---
  /// ZMTP handshake (including security mechanism) failed.
  HandshakeFailed { endpoint: String, error_msg: String },
  /// ZMTP handshake (including security mechanism) succeeded.
  HandshakeSucceeded { endpoint: String },

  // --- Backpressure/Congestion Events ---
  /// The connection's egress pipe has reached its send HWM; the socket is
  /// actively skipping this peer via write-ready skip.
  ConnectionCongested { endpoint: String },
  /// The connection's egress pipe has drained below its send HWM; the peer
  /// re-enters the load-balancer rotation.
  ConnectionUncongested { endpoint: String },
}

/// Strip the internal `#<handle>` uniqueness suffix from a connection URI before
/// exposing it through the monitor channel. Inproc connections use
/// `inproc://name#<handle>` as internal map keys; external observers always see
/// the clean logical URI.
pub(crate) fn clean_endpoint_uri(uri: &str) -> &str {
  match uri.find('#') {
    Some(pos) => &uri[..pos],
    None => uri,
  }
}

// Type alias for the channel sender used for monitor events
pub type MonitorSender = fibre::mpsc::BoundedAsyncSender<SocketEvent>;
// Type alias for the channel receiver used for monitor events
pub type MonitorReceiver = fibre::mpsc::BoundedAsyncReceiver<SocketEvent>;

// Default capacity for monitor channel
pub const DEFAULT_MONITOR_CAPACITY: usize = 100;