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

use crate::codec::{CodecError, Message};
use crate::engine::backend::GenericSocketBackend;
use crate::socket::common::{HasCommon, SocketCommon};
use crate::socket::core::SocketCore;
use crate::{Socket, SocketBackend, SocketOptions, SocketRecv, SocketType, ZmqMessage, ZmqResult};

use flume::Receiver;

/// Gather socket (GATHER). Receives messages fair-queued from connected SCATTER peers.
///
/// Counterpart to [`ScatterSocket`](crate::ScatterSocket).
///
/// 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 GatherSocket {
    core: SocketCore,
    inbound: Receiver<(
        crate::engine::registry::PeerKey,
        Result<Message, CodecError>,
    )>,
}

impl HasCommon for GatherSocket {
    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 GatherSocket {
    type Backend = GenericSocketBackend;

    fn with_options(options: SocketOptions) -> Self {
        let core = SocketCore::new(SocketType::GATHER, options);
        let inbound = core.common.backend.inbound();
        Self { core, inbound }
    }

    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 SocketRecv for GatherSocket {
    async fn recv(&mut self) -> ZmqResult<ZmqMessage> {
        let receive_timeout = self.core.common.backend.socket_options().receive_timeout;
        loop {
            let (_peer_id, message) = self
                .core
                .common
                .backend
                .recv_auto(&self.inbound, receive_timeout)
                .await?;
            if message.len() == 1 {
                return Ok(message);
            }
            // silently drop multipart messages, matching libzmq gather.cpp behaviour
        }
    }
}