rzmq 0.5.21

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

use crate::error::ZmqError;
use crate::socket::types::SocketType;

/// Validates whether two socket types are allowed to connect over `inproc`.
/// Returns `Ok(())` if compatible, or `Err(ZmqError::InvalidSocketType)` if incompatible.
pub(crate) fn validate_socket_compatibility(
  connector: SocketType,
  binder: SocketType,
) -> Result<(), ZmqError> {
  let compatible = match (connector, binder) {
    (SocketType::Push, SocketType::Pull) | (SocketType::Pull, SocketType::Push) => true,
    (SocketType::Pub, SocketType::Sub) | (SocketType::Sub, SocketType::Pub) => true,
    (SocketType::Req, SocketType::Rep) | (SocketType::Rep, SocketType::Req) => true,
    (SocketType::Dealer, SocketType::Router) | (SocketType::Router, SocketType::Dealer) => true,
    _ => false,
  };

  if compatible {
    Ok(())
  } else {
    Err(ZmqError::InvalidSocketType(
      "Incompatible inproc socket patterns",
    ))
  }
}

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

  #[test]
  fn test_valid_compatibility_pairs() {
    assert!(validate_socket_compatibility(SocketType::Push, SocketType::Pull).is_ok());
    assert!(validate_socket_compatibility(SocketType::Pull, SocketType::Push).is_ok());
    assert!(validate_socket_compatibility(SocketType::Pub, SocketType::Sub).is_ok());
    assert!(validate_socket_compatibility(SocketType::Sub, SocketType::Pub).is_ok());
    assert!(validate_socket_compatibility(SocketType::Req, SocketType::Rep).is_ok());
    assert!(validate_socket_compatibility(SocketType::Rep, SocketType::Req).is_ok());
    assert!(validate_socket_compatibility(SocketType::Dealer, SocketType::Router).is_ok());
    assert!(validate_socket_compatibility(SocketType::Router, SocketType::Dealer).is_ok());
  }

  #[test]
  fn test_invalid_compatibility_pairs() {
    assert!(matches!(
      validate_socket_compatibility(SocketType::Push, SocketType::Pub),
      Err(ZmqError::InvalidSocketType(_))
    ));
    assert!(matches!(
      validate_socket_compatibility(SocketType::Req, SocketType::Req),
      Err(ZmqError::InvalidSocketType(_))
    ));
    assert!(matches!(
      validate_socket_compatibility(SocketType::Push, SocketType::Push),
      Err(ZmqError::InvalidSocketType(_))
    ));
    assert!(matches!(
      validate_socket_compatibility(SocketType::Dealer, SocketType::Dealer),
      Err(ZmqError::InvalidSocketType(_))
    ));
    assert!(matches!(
      validate_socket_compatibility(SocketType::Pub, SocketType::Req),
      Err(ZmqError::InvalidSocketType(_))
    ));
  }
}