arzmq/socket/
monitor.rs

1#[cfg(feature = "futures")]
2use core::{pin::Pin, task::Context, task::Poll};
3
4#[cfg(feature = "futures")]
5use async_trait::async_trait;
6#[cfg(feature = "futures")]
7use futures::FutureExt;
8
9use super::{MonitorFlags, MultipartReceiver, RecvFlags, SocketType};
10use crate::{
11    ZmqError, ZmqResult, message::MultipartMessage, sealed, socket::Socket, zmq_sys_crate,
12};
13
14#[derive(Debug, Clone, Copy, PartialEq)]
15/// Errors stemming from [`HandShakeFailedProtocol`]
16///
17/// [`HandShakeFailedProtocol`]: MonitorSocketEvent::HandshakeFailedProtocol
18pub enum HandshakeProtocolError {
19    ZmtpUnspecified,
20    ZmtpUnexpectedCommand,
21    ZmtpInvalidSequence,
22    ZmtpKeyEchange,
23    ZmtpMalformedCommandUnspecified,
24    ZmtpMalformedCommandMessage,
25    ZmtpMalformedCommandHello,
26    ZmtpMalformedCommandInitiate,
27    ZmtpMalformedCommandError,
28    ZmtpMalformedCommandReady,
29    ZmtpMalformedCommandWelcome,
30    ZmtpInvalidMetadata,
31    ZmtpCryptographic,
32    ZmtpMechanismMismatch,
33    ZapUnspecified,
34    ZapMalformedReply,
35    ZapBadRequestId,
36    ZapBadVersion,
37    ZapInvalidStatusCode,
38    ZapInvalidMetadata,
39    UnsupportedError(u32),
40}
41
42impl From<u32> for HandshakeProtocolError {
43    fn from(value: u32) -> Self {
44        match value {
45            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_UNSPECIFIED => Self::ZmtpUnspecified,
46            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_UNEXPECTED_COMMAND => {
47                Self::ZmtpUnexpectedCommand
48            }
49            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_INVALID_SEQUENCE => Self::ZmtpInvalidSequence,
50            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_KEY_EXCHANGE => Self::ZmtpKeyEchange,
51            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_UNSPECIFIED => {
52                Self::ZmtpMalformedCommandUnspecified
53            }
54            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_MESSAGE => {
55                Self::ZmtpMalformedCommandMessage
56            }
57            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_HELLO => {
58                Self::ZmtpMalformedCommandHello
59            }
60            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_INITIATE => {
61                Self::ZmtpMalformedCommandInitiate
62            }
63            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_ERROR => {
64                Self::ZmtpMalformedCommandError
65            }
66            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_READY => {
67                Self::ZmtpMalformedCommandReady
68            }
69            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_WELCOME => {
70                Self::ZmtpMalformedCommandWelcome
71            }
72            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_INVALID_METADATA => Self::ZapInvalidMetadata,
73            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_CRYPTOGRAPHIC => Self::ZmtpCryptographic,
74            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MECHANISM_MISMATCH => {
75                Self::ZmtpMechanismMismatch
76            }
77            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_UNSPECIFIED => Self::ZapUnspecified,
78            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_MALFORMED_REPLY => Self::ZapMalformedReply,
79            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_BAD_REQUEST_ID => Self::ZapBadRequestId,
80            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_BAD_VERSION => Self::ZapBadVersion,
81            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_INVALID_STATUS_CODE => Self::ZapInvalidStatusCode,
82            zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_INVALID_METADATA => Self::ZapInvalidMetadata,
83            other => Self::UnsupportedError(other),
84        }
85    }
86}
87
88#[cfg(test)]
89mod handshake_protocol_error_tests {
90    use rstest::*;
91
92    use super::HandshakeProtocolError;
93    use crate::zmq_sys_crate;
94
95    #[rstest]
96    #[case(
97        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_UNSPECIFIED,
98        HandshakeProtocolError::ZmtpUnspecified
99    )]
100    #[case(
101        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_UNEXPECTED_COMMAND,
102        HandshakeProtocolError::ZmtpUnexpectedCommand
103    )]
104    #[case(
105        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_INVALID_SEQUENCE,
106        HandshakeProtocolError::ZmtpInvalidSequence
107    )]
108    #[case(
109        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_KEY_EXCHANGE,
110        HandshakeProtocolError::ZmtpKeyEchange
111    )]
112    #[case(
113        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_UNSPECIFIED,
114        HandshakeProtocolError::ZmtpMalformedCommandUnspecified
115    )]
116    #[case(
117        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_MESSAGE,
118        HandshakeProtocolError::ZmtpMalformedCommandMessage
119    )]
120    #[case(
121        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_HELLO,
122        HandshakeProtocolError::ZmtpMalformedCommandHello
123    )]
124    #[case(
125        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_INITIATE,
126        HandshakeProtocolError::ZmtpMalformedCommandInitiate
127    )]
128    #[case(
129        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_ERROR,
130        HandshakeProtocolError::ZmtpMalformedCommandError
131    )]
132    #[case(
133        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_READY,
134        HandshakeProtocolError::ZmtpMalformedCommandReady
135    )]
136    #[case(
137        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_WELCOME,
138        HandshakeProtocolError::ZmtpMalformedCommandWelcome
139    )]
140    #[case(
141        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_INVALID_METADATA,
142        HandshakeProtocolError::ZapInvalidMetadata
143    )]
144    #[case(
145        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_CRYPTOGRAPHIC,
146        HandshakeProtocolError::ZmtpCryptographic
147    )]
148    #[case(
149        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_MECHANISM_MISMATCH,
150        HandshakeProtocolError::ZmtpMechanismMismatch
151    )]
152    #[case(
153        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_UNSPECIFIED,
154        HandshakeProtocolError::ZapUnspecified
155    )]
156    #[case(
157        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_MALFORMED_REPLY,
158        HandshakeProtocolError::ZapMalformedReply
159    )]
160    #[case(
161        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_BAD_REQUEST_ID,
162        HandshakeProtocolError::ZapBadRequestId
163    )]
164    #[case(
165        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_BAD_VERSION,
166        HandshakeProtocolError::ZapBadVersion
167    )]
168    #[case(
169        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_INVALID_STATUS_CODE,
170        HandshakeProtocolError::ZapInvalidStatusCode
171    )]
172    #[case(
173        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZAP_INVALID_METADATA,
174        HandshakeProtocolError::ZapInvalidMetadata
175    )]
176    #[case(666, HandshakeProtocolError::UnsupportedError(666))]
177    fn converts_from_raw(#[case] raw_value: u32, #[case] expected: HandshakeProtocolError) {
178        assert_eq!(HandshakeProtocolError::from(raw_value), expected);
179    }
180}
181
182#[derive(Debug, Clone, PartialEq)]
183/// Monitor events that can be received from a monitor socket
184pub enum MonitorSocketEvent {
185    /// The socket has successfully connected to a remote peer. The event value is the file
186    /// descriptor (FD) of the underlying network socket.
187    ///
188    /// <div class="warning">
189    ///
190    /// Warning:
191    ///
192    /// There is no guarantee that the FD is still valid by the time your code receives this
193    /// event.
194    ///
195    /// </div>
196    Connected,
197    /// A connect request on the socket is pending. The event value is unspecified.
198    ConnectDelayed,
199    /// A connect request failed, and is now being retried. The event value is the reconnect
200    /// interval in milliseconds.
201    ///
202    /// Note that the reconnect interval is recalculated at each retry.
203    ConnectRetried(u32),
204    /// The socket was successfully bound to a network interface. The event value is the FD of
205    /// the underlying network socket.
206    ///
207    /// <div class="warning">
208    ///
209    /// Warning:
210    ///
211    /// There is no guarantee that the FD is still valid by the time your code receives this
212    /// event.
213    ///
214    /// </div>
215    Listening,
216    /// The socket could not bind to a given interface. The event value is the errno generated
217    /// by the system bind call.
218    BindFailed,
219    /// The socket has accepted a connection from a remote peer. The event value is the FD of
220    /// the underlying network socket.
221    ///
222    /// <div class="warning">
223    ///
224    /// Warning:
225    ///
226    /// There is no guarantee that the FD is still valid by the time your code receives this
227    /// event.
228    ///
229    /// </div>
230    Accepted,
231    /// The socket has rejected a connection from a remote peer. The event value is the errno
232    /// generated by the accept call.
233    AcceptFailed(ZmqError),
234    /// The socket was closed. The event value is the FD of the (now closed) network socket.
235    Closed,
236    /// The socket close failed. The event value is the errno returned by the system call.
237    ///
238    /// Note that this event occurs only on IPC transports.
239    CloseFailed(ZmqError),
240    /// The socket was disconnected unexpectedly. The event value is the FD of the underlying
241    /// network socket.
242    ///
243    /// <div class="warning">
244    ///
245    /// Warning:
246    ///
247    /// This socket will be closed.
248    ///
249    /// </div>
250    Disconnected,
251    /// Monitoring on this socket ended.
252    MonitorStopped,
253    /// Unspecified error during handshake. The event value is an errno.
254    HandshakeFailedNoDetail(ZmqError),
255    /// The ZMTP security mechanism handshake succeeded. The event value is unspecified.
256    HandshakeSucceeded,
257    /// The ZMTP security mechanism handshake failed due to some mechanism protocol error,
258    /// either between the ZMTP mechanism peers, or between the mechanism server and the ZAP
259    /// handler. This indicates a configuration or implementation error in either peer resp.
260    /// the ZAP handler.
261    HandshakeFailedProtocol(HandshakeProtocolError),
262    /// The ZMTP security mechanism handshake failed due to an authentication failure. The
263    /// event value is the status code returned by the ZAP handler (i.e. `300`, `400` or `500`).
264    HandshakeFailedAuth(u32),
265    UnSupported(MonitorFlags, u32),
266}
267
268impl TryFrom<MultipartMessage> for MonitorSocketEvent {
269    type Error = ZmqError;
270
271    fn try_from(zmq_msgs: MultipartMessage) -> Result<Self, Self::Error> {
272        if zmq_msgs.len() != 2 {
273            return Err(ZmqError::InvalidArgument);
274        }
275
276        let Some(first_msg) = zmq_msgs.get(0) else {
277            unreachable!();
278        };
279
280        if first_msg.len() != 6 {
281            return Err(ZmqError::InvalidArgument);
282        }
283
284        let Some(event_id) = first_msg
285            .bytes()
286            .first_chunk::<2>()
287            .map(|raw_event_id| u16::from_le_bytes(*raw_event_id))
288            .map(MonitorFlags::from)
289        else {
290            unreachable!();
291        };
292
293        let Some(event_value) = first_msg
294            .bytes()
295            .last_chunk::<4>()
296            .map(|raw_event_value| u32::from_le_bytes(*raw_event_value))
297        else {
298            unreachable!();
299        };
300
301        match event_id {
302            MonitorFlags::Connected => Ok(Self::Connected),
303            MonitorFlags::ConnectDelayed => Ok(Self::ConnectDelayed),
304            MonitorFlags::ConnectRetried => Ok(Self::ConnectRetried(event_value)),
305            MonitorFlags::Listening => Ok(Self::Listening),
306            MonitorFlags::Accepted => Ok(Self::Accepted),
307            MonitorFlags::AcceptFailed => {
308                Ok(Self::AcceptFailed(ZmqError::from(event_value as i32)))
309            }
310            MonitorFlags::Closed => Ok(Self::Closed),
311            MonitorFlags::CloseFailed => Ok(Self::CloseFailed(ZmqError::from(event_value as i32))),
312            MonitorFlags::Disconnected => Ok(Self::Disconnected),
313            MonitorFlags::MonitorStopped => Ok(Self::MonitorStopped),
314            MonitorFlags::HandshakeFailedNoDetail => Ok(Self::HandshakeFailedNoDetail(
315                ZmqError::from(event_value as i32),
316            )),
317            MonitorFlags::HandshakeSucceeded => Ok(Self::HandshakeSucceeded),
318            MonitorFlags::HandshakeFailedProtocol => {
319                Ok(Self::HandshakeFailedProtocol(event_value.into()))
320            }
321            MonitorFlags::HandshakeFailedAuth => Ok(Self::HandshakeFailedAuth(event_value)),
322            event_id => Ok(Self::UnSupported(event_id, event_value)),
323        }
324    }
325}
326
327#[cfg(test)]
328mod monitor_socket_event_tests {
329    use rstest::*;
330
331    use super::{HandshakeProtocolError, MonitorSocketEvent};
332    use crate::{
333        prelude::{MonitorFlags, MultipartMessage, ZmqError, ZmqResult},
334        zmq_sys_crate,
335    };
336
337    #[rstest]
338    #[case(MonitorFlags::Connected, 0, Ok(MonitorSocketEvent::Connected))]
339    #[case(
340        MonitorFlags::ConnectDelayed,
341        0,
342        Ok(MonitorSocketEvent::ConnectDelayed)
343    )]
344    #[case(
345        MonitorFlags::ConnectRetried,
346        42,
347        Ok(MonitorSocketEvent::ConnectRetried(42))
348    )]
349    #[case(MonitorFlags::Listening, 0, Ok(MonitorSocketEvent::Listening))]
350    #[case(MonitorFlags::Accepted, 0, Ok(MonitorSocketEvent::Accepted))]
351    #[case(
352        MonitorFlags::AcceptFailed,
353        14,
354        Ok(MonitorSocketEvent::AcceptFailed(ZmqError::ContextInvalid))
355    )]
356    #[case(MonitorFlags::Closed, 0, Ok(MonitorSocketEvent::Closed))]
357    #[case(
358        MonitorFlags::CloseFailed,
359        14,
360        Ok(MonitorSocketEvent::CloseFailed(ZmqError::ContextInvalid))
361    )]
362    #[case(MonitorFlags::Disconnected, 0, Ok(MonitorSocketEvent::Disconnected))]
363    #[case(
364        MonitorFlags::MonitorStopped,
365        0,
366        Ok(MonitorSocketEvent::MonitorStopped)
367    )]
368    #[case(
369        MonitorFlags::HandshakeFailedNoDetail,
370        14,
371        Ok(MonitorSocketEvent::HandshakeFailedNoDetail(ZmqError::ContextInvalid))
372    )]
373    #[case(
374        MonitorFlags::HandshakeSucceeded,
375        0,
376        Ok(MonitorSocketEvent::HandshakeSucceeded)
377    )]
378    #[case(
379        MonitorFlags::HandshakeFailedProtocol,
380        zmq_sys_crate::ZMQ_PROTOCOL_ERROR_ZMTP_UNSPECIFIED,
381        Ok(MonitorSocketEvent::HandshakeFailedProtocol(HandshakeProtocolError::ZmtpUnspecified))
382    )]
383    #[case(
384        MonitorFlags::HandshakeFailedAuth,
385        404,
386        Ok(MonitorSocketEvent::HandshakeFailedAuth(404))
387    )]
388    #[case(
389        MonitorFlags::HandshakeFailedAuth | MonitorFlags::Connected,
390        42,
391        Ok(MonitorSocketEvent::UnSupported(MonitorFlags::HandshakeFailedAuth | MonitorFlags::Connected, 42))
392    )]
393    fn try_from_multipart_succeeds(
394        #[case] upper_chunk: MonitorFlags,
395        #[case] lower_chunk: u32,
396        #[case] expected: ZmqResult<MonitorSocketEvent>,
397    ) {
398        let mut first = upper_chunk.bits().to_le_bytes().to_vec();
399        first.extend(lower_chunk.to_le_bytes());
400        let multipart: MultipartMessage = vec![first.into(), vec![].into()].into();
401
402        assert_eq!(MonitorSocketEvent::try_from(multipart), expected);
403    }
404
405    #[test]
406    fn try_from_mutipart_with_too_few_parts() {
407        let multipart: MultipartMessage = vec!["asdf".into()].into();
408        let result = MonitorSocketEvent::try_from(multipart);
409
410        assert!(result.is_err_and(|err| err == ZmqError::InvalidArgument));
411    }
412
413    #[test]
414    fn try_from_mutipart_with_too_many_parts() {
415        let multipart: MultipartMessage = vec!["asdf".into(), "asdf".into(), "asdf".into()].into();
416        let result = MonitorSocketEvent::try_from(multipart);
417
418        assert!(result.is_err_and(|err| err == ZmqError::InvalidArgument));
419    }
420
421    #[test]
422    fn try_from_mutipart_with_too_short_first_part() {
423        let multipart: MultipartMessage = vec![vec![1, 2, 3, 4, 5].into(), "asdf".into()].into();
424        let result = MonitorSocketEvent::try_from(multipart);
425
426        assert!(result.is_err_and(|err| err == ZmqError::InvalidArgument));
427    }
428
429    #[test]
430    fn try_from_mutipart_with_too_long_first_part() {
431        let multipart: MultipartMessage =
432            vec![vec![1, 2, 3, 4, 5, 6, 7].into(), "asdf".into()].into();
433        let result = MonitorSocketEvent::try_from(multipart);
434
435        assert!(result.is_err_and(|err| err == ZmqError::InvalidArgument));
436    }
437}
438
439/// # A monitor socket `ZMQ_PAIR`
440pub type MonitorSocket = Socket<Monitor>;
441
442pub struct Monitor {}
443
444impl sealed::ReceiverFlag for Monitor {}
445
446unsafe impl Sync for Socket<Monitor> {}
447unsafe impl Send for Socket<Monitor> {}
448
449impl MultipartReceiver for Socket<Monitor> {}
450
451impl sealed::SocketType for Monitor {
452    fn raw_socket_type() -> SocketType {
453        SocketType::Pair
454    }
455}
456
457impl Socket<Monitor> {}
458
459#[cfg_attr(feature = "futures", async_trait)]
460/// Trait for receiving [`MonitorSocketEvent`] from a monitor socket
461///
462/// [`MonitorSocketEvent`]: MonitorSocketEvent
463pub trait MonitorReceiver {
464    fn recv_monitor_event(&self) -> ZmqResult<MonitorSocketEvent>;
465
466    #[cfg(feature = "futures")]
467    async fn recv_monitor_event_async(&self) -> Option<MonitorSocketEvent>;
468}
469
470#[cfg_attr(feature = "futures", async_trait)]
471impl MonitorReceiver for MonitorSocket {
472    fn recv_monitor_event(&self) -> ZmqResult<MonitorSocketEvent> {
473        self.recv_multipart(RecvFlags::DONT_WAIT)
474            .and_then(MonitorSocketEvent::try_from)
475    }
476
477    #[cfg(feature = "futures")]
478    async fn recv_monitor_event_async(&self) -> Option<MonitorSocketEvent> {
479        MonitorSocketEventFuture { receiver: self }.now_or_never()
480    }
481}
482
483#[cfg(feature = "futures")]
484struct MonitorSocketEventFuture<'a> {
485    receiver: &'a MonitorSocket,
486}
487
488#[cfg(feature = "futures")]
489impl Future for MonitorSocketEventFuture<'_> {
490    type Output = MonitorSocketEvent;
491
492    fn poll(self: Pin<&mut Self>, _ctx: &mut Context<'_>) -> Poll<Self::Output> {
493        match self.receiver.recv_monitor_event() {
494            Ok(event) => Poll::Ready(event),
495            _ => Poll::Pending,
496        }
497    }
498}
arzmq/socket/monitor.rs

arzmq/socket/
monitor.rs
