ruststream_zeromq/
endpoint.rs

1//! [`ZmqEndpoint`]: an address plus the explicit bind-or-connect role.
2//!
3//! There is no server in the middle, so which side listens is a deployment decision, not a
4//! property of the transport; the endpoint states it.
5
6use crate::error::ZmqError;
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub(crate) enum Role {
10    Bind,
11    Connect,
12}
13
14/// An address (`tcp://...` or `ipc://...`) with an explicit listening role.
15///
16/// # Examples
17///
18/// ```
19/// use ruststream_zeromq::ZmqEndpoint;
20///
21/// let listener = ZmqEndpoint::bind("tcp://0.0.0.0:5555");
22/// let dialer = ZmqEndpoint::connect("tcp://ml:5555");
23/// let local = ZmqEndpoint::bind("ipc:///tmp/orders");
24/// # let _ = (listener, dialer, local);
25/// ```
26#[derive(Debug, Clone, PartialEq, Eq)]
27#[must_use]
28pub struct ZmqEndpoint {
29    pub(crate) address: String,
30    pub(crate) role: Role,
31}
32
33impl ZmqEndpoint {
34    /// This process listens on `address`.
35    pub fn bind(address: impl Into<String>) -> Self {
36        Self {
37            address: address.into(),
38            role: Role::Bind,
39        }
40    }
41
42    /// This process dials out to `address`.
43    pub fn connect(address: impl Into<String>) -> Self {
44        Self {
45            address: address.into(),
46            role: Role::Connect,
47        }
48    }
49
50    /// The address string.
51    #[must_use]
52    pub fn address(&self) -> &str {
53        &self.address
54    }
55
56    /// Rejects endpoints the implementation cannot serve, before any I/O.
57    pub(crate) fn validate(&self) -> Result<(), ZmqError> {
58        if self.address.starts_with("tcp://") || self.address.starts_with("ipc://") {
59            Ok(())
60        } else {
61            Err(ZmqError::Invalid(format!(
62                "'{}' must use the tcp:// or ipc:// transport",
63                self.address
64            )))
65        }
66    }
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72
73    #[test]
74    fn unsupported_transports_are_rejected_before_io() {
75        assert!(ZmqEndpoint::bind("inproc://x").validate().is_err());
76        assert!(ZmqEndpoint::connect("udp://x:1").validate().is_err());
77        assert!(ZmqEndpoint::bind("tcp://0.0.0.0:5555").validate().is_ok());
78        assert!(ZmqEndpoint::bind("ipc:///tmp/x").validate().is_ok());
79    }
80}
ruststream_zeromq/endpoint.rs

ruststream_zeromq/
endpoint.rs
