async_zmq/
pair.rs

1//! PAIR socket module of Exclusive pair pattern in ZMQ
2//!
3//! Use the [`pair`] function to instantiate a pair socket and use methods from
4//! the [`Sink`]/[`SinkExt`] and [`Stream`]/[`StreamExt`] traits.
5//!
6//! A pair socket must be paired with another pair socket.
7//!
8//! # Example
9//!
10//! ```no_run
11//! use async_zmq::{Result, SinkExt};
12//!
13//! #[async_std::main]
14//! async fn main() -> Result<()> {
15//!     let mut zmq = async_zmq::pair("tcp://127.0.0.1:5555")?.bind()?;
16//!
17//!     zmq.send(vec!["broadcast message"].into()).await?;
18//!     Ok(())
19//! }
20//! ```
21//!
22//! [`pair`]: fn.pair.html
23//! [`Sink`]: ../trait.Sink.html
24//! [`SinkExt`]: ../trait.SinkExt.html
25//! [`Stream`]: ../trait.Stream.html
26//! [`StreamExt`]: ../trait.StreamExt.html
27
28use std::pin::Pin;
29use std::task::{Context, Poll};
30
31use zmq::{Message, SocketType};
32
33use crate::{
34    reactor::{AsRawSocket, ZmqSocket},
35    socket::{Broker, Multipart, MultipartIter, SocketBuilder},
36    RecvError, SendError, Sink, SocketError, Stream,
37};
38
39/// Create a ZMQ socket with PAIR type
40pub fn pair<I: Iterator<Item = T> + Unpin, T: Into<Message>>(
41    endpoint: &str,
42) -> Result<SocketBuilder<'_, Pair<I, T>>, SocketError> {
43    Ok(SocketBuilder::new(SocketType::PAIR, endpoint))
44}
45
46/// The async wrapper of ZMQ socket with PAIR type
47pub struct Pair<I: Iterator<Item = T> + Unpin, T: Into<Message>>(Broker<I, T>);
48
49impl<I: Iterator<Item = T> + Unpin, T: Into<Message>> Pair<I, T> {
50    /// Represent as `Socket` from zmq crate in case you want to call its methods.
51    pub fn as_raw_socket(&self) -> &zmq::Socket {
52        self.0.socket.as_socket()
53    }
54}
55
56impl<I: Iterator<Item = T> + Unpin, T: Into<Message>> Sink<MultipartIter<I, T>> for Pair<I, T> {
57    type Error = SendError;
58
59    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
60        Sink::poll_ready(Pin::new(&mut self.get_mut().0), cx)
61            .map(|result| result.map_err(Into::into))
62    }
63
64    fn start_send(self: Pin<&mut Self>, item: MultipartIter<I, T>) -> Result<(), Self::Error> {
65        Pin::new(&mut self.get_mut().0)
66            .start_send(item)
67            .map_err(Into::into)
68    }
69
70    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
71        Sink::poll_flush(Pin::new(&mut self.get_mut().0), cx)
72            .map(|result| result.map_err(Into::into))
73    }
74
75    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
76        Sink::poll_close(Pin::new(&mut self.get_mut().0), cx)
77            .map(|result| result.map_err(Into::into))
78    }
79}
80
81impl<I: Iterator<Item = T> + Unpin, T: Into<Message>> Stream for Pair<I, T> {
82    type Item = Result<Multipart, RecvError>;
83
84    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
85        Pin::new(&mut self.get_mut().0)
86            .poll_next(cx)
87            .map(|poll| poll.map(|result| result.map_err(Into::into)))
88    }
89}
90
91impl<I: Iterator<Item = T> + Unpin, T: Into<Message>> From<zmq::Socket> for Pair<I, T> {
92    fn from(socket: zmq::Socket) -> Self {
93        Self(Broker {
94            socket: ZmqSocket::from(socket),
95            buffer: None,
96        })
97    }
98}
async_zmq/pair.rs

async_zmq/
pair.rs
