celerity 0.1.1

Pure Rust sans-IO ZMTP 3.1 messaging core with Tokio TCP and Unix socket transports.
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

//! Public API smoke tests.

use bytes::Bytes;
use celerity::{MetadataMap, ProtocolError};

#[cfg(feature = "tokio")]
use celerity::io::{Endpoint, TokioCelerityError, TransportKind};

fn ok<T, E: core::fmt::Debug>(result: Result<T, E>) -> T {
    match result {
        Ok(value) => value,
        Err(err) => panic!("expected Ok(..), got Err({err:?})"),
    }
}

fn err<T, E>(result: Result<T, E>) -> E {
    match result {
        Ok(_) => panic!("expected Err(..), got Ok(..)"),
        Err(err) => err,
    }
}

#[test]
fn metadata_map_replaces_keys_case_insensitively() {
    let mut metadata = MetadataMap::new();
    ok(metadata.insert("Identity", "first"));
    ok(metadata.insert("identity", "second"));

    assert_eq!(metadata.len(), 1);
    assert_eq!(
        metadata.get("IDENTITY").cloned(),
        Some(Bytes::from_static(b"second"))
    );
}

#[test]
fn metadata_map_rejects_invalid_names() {
    assert_eq!(
        err(MetadataMap::new().insert("", "value")),
        ProtocolError::InvalidMetadataName(Bytes::new())
    );
    assert_eq!(
        err(MetadataMap::new().insert("bad name", "value")),
        ProtocolError::InvalidMetadataName(Bytes::from_static(b"bad name"))
    );

    let long_name = vec![b'a'; 256];
    assert_eq!(
        err(MetadataMap::new().insert(Bytes::from(long_name.clone()), "value")),
        ProtocolError::InvalidMetadataName(Bytes::from(long_name))
    );
}

#[cfg(feature = "tokio")]
#[test]
fn endpoint_parsing_accepts_tcp_and_ipc_inputs() {
    assert_eq!(
        ok(Endpoint::parse("tcp://127.0.0.1:5555")),
        Endpoint::Tcp("127.0.0.1:5555".to_owned())
    );
    assert_eq!(
        ok(Endpoint::parse("127.0.0.1:5555")),
        Endpoint::Tcp("127.0.0.1:5555".to_owned())
    );

    #[cfg(unix)]
    assert_eq!(
        ok(Endpoint::parse("ipc:///tmp/celerity-test.sock")).transport_kind(),
        TransportKind::Ipc
    );
}

#[cfg(feature = "tokio")]
#[test]
fn endpoint_parsing_rejects_invalid_inputs() {
    assert!(matches!(
        err(Endpoint::parse("")),
        TokioCelerityError::InvalidEndpoint(endpoint) if endpoint.is_empty()
    ));
    assert!(matches!(
        err(Endpoint::parse("tcp://")),
        TokioCelerityError::InvalidEndpoint(endpoint) if endpoint == "tcp://"
    ));
    assert!(matches!(
        err(Endpoint::parse("udp://127.0.0.1:5555")),
        TokioCelerityError::UnsupportedEndpoint(endpoint)
            if endpoint == "udp://127.0.0.1:5555"
    ));

    #[cfg(unix)]
    assert!(matches!(
        err(Endpoint::parse("ipc://relative.sock")),
        TokioCelerityError::InvalidEndpoint(endpoint) if endpoint == "ipc://relative.sock"
    ));
}