rustzmq2 0.1.0

A native async Rust implementation of ZeroMQ
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242

//! ZMTP greeting + READY command exchange over a `FramedIo`.
//!
//! Implements the wire-level handshake from [ZMTP RFC 23](https://rfc.zeromq.org/spec/23/):
//! swap `ZmqGreeting`, negotiate version, then swap `READY` commands carrying
//! socket-type + metadata properties. Security-mechanism handshakes (PLAIN,
//! CURVE) run *between* greeting and READY and live in `crate::mechanism`.

#[cfg(all(feature = "tokio", any(feature = "tcp", test)))]
use crate::codec::ZmtpVersion;
#[cfg(any(
    all(feature = "tokio", any(feature = "tcp", test)),
    feature = "tcp",
    all(feature = "ipc", target_family = "unix")
))]
use crate::codec::{CodecError, FramedIo, Message, ZmqGreeting};
#[cfg(any(feature = "tcp", all(feature = "ipc", target_family = "unix")))]
use crate::codec::{CodecResult, ZmqCommand, ZmqCommandName};
#[cfg(any(feature = "tcp", all(feature = "ipc", target_family = "unix")))]
use crate::peer_identity::PeerIdentity;
#[cfg(any(feature = "tcp", all(feature = "ipc", target_family = "unix")))]
use crate::SocketOptions;
#[cfg(any(feature = "tcp", all(feature = "ipc", target_family = "unix")))]
use crate::SocketType;
#[cfg(any(
    all(feature = "tokio", any(feature = "tcp", test)),
    feature = "tcp",
    all(feature = "ipc", target_family = "unix")
))]
use crate::{ZmqError, ZmqResult};

#[cfg(any(feature = "tcp", all(feature = "ipc", target_family = "unix")))]
use std::collections::HashMap;
#[cfg(any(feature = "tcp", all(feature = "ipc", target_family = "unix")))]
use std::convert::{TryFrom, TryInto};

#[cfg(any(feature = "tcp", all(feature = "ipc", target_family = "unix")))]
use bytes::Bytes;
#[cfg(any(
    all(feature = "tokio", any(feature = "tcp", test)),
    feature = "tcp",
    all(feature = "ipc", target_family = "unix")
))]
use futures::{Sink, SinkExt, Stream, StreamExt};

/// Given the result of the greetings exchange, determines the version of the
/// ZMTP protocol that should be used for communication with the peer according
/// to [ZeroMQ RFC 23](https://rfc.zeromq.org/spec/23/#version-negotiation).
#[cfg(all(feature = "tokio", any(feature = "tcp", test)))]
pub(crate) fn negotiate_version(greeting: Message) -> ZmqResult<ZmtpVersion> {
    let my_version = ZmqGreeting::default().version;

    match greeting {
        Message::Greeting(peer) => {
            if peer.version >= my_version {
                // A peer MUST accept higher protocol versions as valid. That is,
                // a ZMTP peer MUST accept protocol versions greater or equal to 3.0.
                // This allows future implementations to safely interoperate with
                // current implementations.
                //
                // A peer SHALL always use its own protocol (including framing)
                // when talking to an equal or higher protocol peer.
                Ok(my_version)
            } else {
                // Per RFC 23: a peer MAY downgrade to talk to an older peer, but if it
                // cannot, it MUST close the connection. We do not support downgrading.
                Err(ZmqError::UnsupportedVersion(peer.version))
            }
        }
        _ => Err(ZmqError::Other("Failed Greeting exchange".into())),
    }
}

#[cfg(all(feature = "tokio", any(feature = "tcp", test)))]
pub(crate) async fn greet_exchange<R, W>(raw_socket: &mut FramedIo<R, W>) -> ZmqResult<ZmtpVersion>
where
    R: Stream<Item = Result<Message, CodecError>> + Unpin,
    W: Sink<Message, Error = CodecError> + Unpin,
{
    raw_socket
        .write_half
        .send(Message::Greeting(ZmqGreeting::default()))
        .await?;

    let greeting = match raw_socket.read_half.next().await {
        Some(message) => message?,
        None => return Err(ZmqError::Other("Failed Greeting exchange".into())),
    };
    negotiate_version(greeting)
}

/// Like `greet_exchange` but returns the full peer `ZmqGreeting` so that the
/// caller can inspect the peer's mechanism and server flag.
#[cfg(any(feature = "tcp", all(feature = "ipc", target_family = "unix")))]
pub(crate) async fn greet_exchange_full<R, W>(
    raw_socket: &mut FramedIo<R, W>,
    opts: &SocketOptions,
) -> ZmqResult<ZmqGreeting>
where
    R: Stream<Item = Result<Message, CodecError>> + Unpin,
    W: Sink<Message, Error = CodecError> + Unpin,
{
    raw_socket
        .write_half
        .send(Message::Greeting(ZmqGreeting::from_options(opts)))
        .await?;

    match raw_socket.read_half.next().await {
        Some(Ok(Message::Greeting(peer))) => {
            let my_version = ZmqGreeting::default().version;
            if peer.version < my_version {
                return Err(ZmqError::UnsupportedVersion(peer.version));
            }
            Ok(peer)
        }
        Some(Ok(_)) | None => Err(ZmqError::Other("Failed Greeting exchange".into())),
        Some(Err(e)) => Err(e.into()),
    }
}

#[cfg(any(feature = "tcp", all(feature = "ipc", target_family = "unix")))]
pub(crate) async fn ready_exchange<R, W>(
    raw_socket: &mut FramedIo<R, W>,
    socket_type: SocketType,
    props: Option<HashMap<String, Bytes>>,
) -> ZmqResult<PeerIdentity>
where
    R: Stream<Item = Result<Message, CodecError>> + Unpin,
    W: Sink<Message, Error = CodecError> + Unpin,
{
    let mut ready = ZmqCommand::ready(socket_type);
    if let Some(props) = props {
        ready.add_properties(props);
    }
    raw_socket.write_half.send(Message::Command(ready)).await?;

    let ready_repl: Option<CodecResult<Message>> = raw_socket.read_half.next().await;
    match ready_repl {
        Some(Ok(Message::Command(command))) => match command.name {
            ZmqCommandName::READY => {
                let other_sock_type = match command.properties.get("Socket-Type") {
                    Some(s) => SocketType::try_from(&s[..])?,
                    None => Err(ZmqError::Other("Failed to parse other socket type".into()))?,
                };

                let peer_id = command
                    .properties
                    .get("Identity")
                    .map(|x| x.clone().try_into())
                    .transpose()?
                    .unwrap_or_default();

                if socket_type.compatible(other_sock_type) {
                    Ok(peer_id)
                } else {
                    Err(ZmqError::IncompatiblePeer)
                }
            }
        },
        // After a CURVE/PLAIN handshake the codec routes READY as SecurityRaw.
        // Re-parse it as a ZmqCommand so the rest of this function stays the same.
        Some(Ok(Message::SecurityRaw(raw))) => {
            let command = ZmqCommand::try_from(raw).map_err(ZmqError::from)?;
            match command.name {
                ZmqCommandName::READY => {
                    let other_sock_type = match command.properties.get("Socket-Type") {
                        Some(s) => SocketType::try_from(&s[..])?,
                        None => {
                            return Err(ZmqError::Other("Failed to parse other socket type".into()))
                        }
                    };
                    let peer_id = command
                        .properties
                        .get("Identity")
                        .map(|x| x.clone().try_into())
                        .transpose()?
                        .unwrap_or_default();
                    if socket_type.compatible(other_sock_type) {
                        Ok(peer_id)
                    } else {
                        Err(ZmqError::Other(
                            "Provided sockets combination is not compatible".into(),
                        ))
                    }
                }
            }
        }
        Some(Ok(_)) => Err(ZmqError::Other("Failed to confirm ready state".into())),
        Some(Err(e)) => Err(e.into()),
        None => Err(ZmqError::Other("No reply from server".into())),
    }
}

#[cfg(all(test, feature = "tokio"))]
mod tests {
    use super::*;
    use crate::codec::mechanism::ZmqMechanism;
    use crate::message::ZmqMessage;

    fn new_greeting(version: ZmtpVersion) -> Message {
        Message::Greeting(ZmqGreeting {
            version,
            mechanism: ZmqMechanism::PLAIN,
            as_server: false,
        })
    }

    #[test]
    fn negotiate_version_peer_is_using_the_same_version() {
        let peer_version = ZmqGreeting::default().version;
        let expected = ZmqGreeting::default().version;
        let actual = negotiate_version(new_greeting(peer_version)).unwrap();
        assert_eq!(actual, expected);
    }

    #[test]
    fn negotiate_version_peer_is_using_a_newer_version() {
        let peer_version = (3, 1);
        let expected = ZmqGreeting::default().version;
        let actual = negotiate_version(new_greeting(peer_version)).unwrap();
        assert_eq!(actual, expected);
    }

    #[test]
    fn negotiate_version_peer_is_using_an_older_version() {
        let peer_version = (2, 1);
        let actual = negotiate_version(new_greeting(peer_version));
        match actual {
            Err(ZmqError::UnsupportedVersion(version)) => assert_eq!(version, peer_version),
            _ => panic!("Unexpected result"),
        }
    }

    #[test]
    fn negotiate_version_invalid_greeting() {
        let message = Message::Message(ZmqMessage::from(""));
        let actual = negotiate_version(message);
        match actual {
            Err(ZmqError::Other(_)) => {}
            _ => panic!("Unexpected result"),
        }
    }
}