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

use crate::engine::backend::GenericSocketBackend;
use crate::socket::common::{HasCommon, SocketCommon};
use crate::socket::core::SocketCore;
use crate::{
    Socket, SocketBackend, SocketOptions, SocketSend, SocketType, ZmqError, ZmqMessage, ZmqResult,
};

/// Scatter socket (SCATTER). Distributes messages round-robin to connected GATHER peers.
///
/// Counterpart to [`GatherSocket`](crate::GatherSocket). Similar to PUSH/PULL but
/// intended for symmetric scatter/gather pipelines.
///
/// See [`zmq_socket(3)`](https://libzmq.readthedocs.io/en/latest/zmq_socket.html) —
/// SCATTER/GATHER are the single-part analogues of PUSH/PULL.
pub struct ScatterSocket {
    core: SocketCore,
}

impl HasCommon for ScatterSocket {
    type Backend = GenericSocketBackend;
    fn common(&self) -> &SocketCommon<Self::Backend> {
        &self.core.common
    }
    fn common_mut(&mut self) -> &mut SocketCommon<Self::Backend> {
        &mut self.core.common
    }
}

impl Socket for ScatterSocket {
    type Backend = GenericSocketBackend;

    fn with_options(options: SocketOptions) -> Self {
        Self {
            core: SocketCore::new(SocketType::SCATTER, options),
        }
    }

    async fn connect<E>(&mut self, endpoint: E) -> ZmqResult<()>
    where
        E: TryInto<crate::endpoint::Endpoint> + Send,
        E::Error: Into<crate::ZmqError>,
    {
        let endpoint = endpoint.try_into().map_err(Into::into)?;
        self.core.connect_endpoint(endpoint).await
    }

    async fn linger_drain(&mut self) {
        self.core.linger_drain().await;
    }
}

impl SocketSend for ScatterSocket {
    async fn send(&mut self, message: impl Into<ZmqMessage> + Send) -> ZmqResult<()> {
        let message = message.into();
        if message.len() > 1 {
            return Err(ZmqError::Socket(
                "SCATTER does not support multipart messages".into(),
            ));
        }
        let send_timeout = self.core.common.backend.socket_options().send_timeout;
        self.core
            .common
            .backend
            .send_round_robin_timed(message, send_timeout)
            .await?;
        self.core.after_send().await;
        Ok(())
    }
}