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

//! Backend-side traits: the I/O surface sockets implement.
//!
//! `SocketBackend` and `MultiPeerBackend` decouple the socket façade
//! (Socket / `SocketSend` / `SocketRecv`) from the engine-side registry and
//! per-peer I/O tasks that actually shuttle bytes on/off the wire.
//!
//! ## Reference backends
//!
//! - [`GenericSocketBackend`](crate::engine::backend::GenericSocketBackend)
//!   is used by the 10 kinds with uniform send/recv semantics (PUSH,
//!   PULL, DEALER, ROUTER, REQ, REP, PAIR, CHANNEL, SCATTER, GATHER).
//!   Its state fits the `SocketCore` shape: one backend Arc, the
//!   registry, per-engine outbound queues.
//! - `PubSocketBackend` and `XPubSocketBackend` share a
//!   [`TopicRouter`](crate::socket::topic_router) for per-peer
//!   topic-prefix matching. PUB drops sub-change events; XPUB surfaces
//!   them via `recv` (filtered by `xpub_verbose`).
//! - `SubSocketBackend` is the odd one out: SUB keeps state on the
//!   *outbound* side — an auto-replay list of the user's SUBSCRIBE
//!   frames that gets pushed to every new or reconnecting peer. It does
//!   not use `TopicRouter`.

use crate::codec::{CodecError, FramedIo, IntoEngineWriter, Message};
use crate::endpoint::Endpoint;
#[allow(unused_imports)]
use crate::io_compat::AsyncVectoredWrite;
use crate::socket::{SocketEvent, SocketOptions, SocketType};
use crate::PeerIdentity;
use crate::{ZmqMessage, ZmqResult};

use futures::channel::mpsc;
use futures::{Sink, Stream};
use parking_lot::Mutex;

use std::sync::Arc;

/// Backend trait for sockets that manage multiple peer connections.
pub trait MultiPeerBackend: SocketBackend {
    fn peer_connected<R, W>(
        self: Arc<Self>,
        peer_id: &PeerIdentity,
        io: FramedIo<R, W>,
        endpoint: Option<Endpoint>,
    ) -> impl std::future::Future<Output = ()> + Send
    where
        R: Stream<Item = Result<Message, CodecError>> + Unpin + Send + 'static,
        W: Sink<Message, Error = CodecError> + Unpin + Send + IntoEngineWriter + 'static,
        W::Writer: Send + 'static;

    #[cfg(feature = "inproc")]
    #[allow(private_interfaces)]
    fn peer_connected_inproc(
        self: Arc<Self>,
        peer_id: &PeerIdentity,
        peer: crate::transport::inproc::InprocPeer,
        endpoint: Option<Endpoint>,
    ) -> impl std::future::Future<Output = ZmqResult<()>> + Send;

    fn peer_disconnected(&self, peer_id: &PeerIdentity);

    /// Hook fired by the reconnect task after a successful reconnect.
    /// Default no-op. SUB overrides it to deliver `hiccup_msg` via the
    /// inbound channel. Not called on the first connect.
    fn on_reconnect(&self, _peer_id: &PeerIdentity) {}
}

/// Low-level interface implemented by every socket backend.
pub trait SocketBackend: Send + Sync {
    fn socket_type(&self) -> SocketType;
    fn socket_options(&self) -> &SocketOptions;
    fn shutdown(&self);
    fn monitor(&self) -> &Mutex<Option<mpsc::Sender<SocketEvent>>>;
}

/// Receive half of a ZMQ socket.
pub trait SocketRecv {
    /// Receive the next message, blocking until one arrives.
    ///
    /// See [`zmq_recv(3)`](https://libzmq.readthedocs.io/en/latest/zmq_recv.html).
    fn recv(&mut self) -> impl std::future::Future<Output = ZmqResult<ZmqMessage>> + Send;
}

/// Send half of a ZMQ socket.
pub trait SocketSend {
    /// Send a message, blocking until it is accepted by the outbound queue.
    ///
    /// Accepts anything that converts into a [`ZmqMessage`] — `&str`,
    /// `String`, `Vec<u8>`, `Bytes`, `&[u8]`, or a `ZmqMessage` you built
    /// up explicitly.
    ///
    /// See [`zmq_send(3)`](https://libzmq.readthedocs.io/en/latest/zmq_send.html).
    fn send(
        &mut self,
        message: impl Into<ZmqMessage> + Send,
    ) -> impl std::future::Future<Output = ZmqResult<()>> + Send;
}

/// Marker trait expressing the fact that only certain types of sockets
/// may be used in the [`proxy`](crate::proxy) function as a capture
/// parameter.
pub trait CaptureSocket: SocketSend {}