Struct PairSocket 

pub struct PairSocket<S = TcpStream>
where
    S: AsyncRead + AsyncWrite + Unpin,
{ /* private fields */ }

PAIR socket for exclusive peer-to-peer communication.

PAIR sockets connect exactly two endpoints and provide bidirectional message passing without any routing or filtering logic.

Implementations

impl<S> PairSocket<S>

where S: AsyncRead + AsyncWrite + Unpin,

pub async fn new(stream: S) -> Result<PairSocket<S>, Error>

Create a new PAIR socket from a stream with default buffer configuration.

pub async fn with_options( stream: S, options: SocketOptions, ) -> Result<PairSocket<S>, Error>

Create a new PAIR socket with custom buffer configuration and socket options.

pub async fn send(&mut self, msg: Vec<Bytes>) -> Result<(), Error>

Send a message to the paired socket.

Errors

Returns an error if the socket is poisoned, disconnected, or if the write fails.

pub async fn recv(&mut self) -> Result<Option<Vec<Bytes>>, Error>

Receive a message from the paired socket.

Returns Ok(Some(msg)) if a message was received, Ok(None) if the connection was closed, or an error.

pub async fn close(self) -> Result<(), Error>

Close the socket gracefully by shutting down the underlying stream.

pub const fn options(&self) -> &SocketOptions

Get a reference to the socket options.

pub fn options_mut(&mut self) -> &mut SocketOptions

Get a mutable reference to the socket options.

pub fn set_options(&mut self, options: SocketOptions)

Set socket options (builder-style).

pub const fn socket_type(&self) -> SocketType

Get the socket type.

ZeroMQ Compatibility

Corresponds to ZMQ_TYPE (16) option.

pub fn last_endpoint(&self) -> Option<&Endpoint>

Get the endpoint this socket is connected/bound to, if available.

Returns None if the socket was created from a raw stream.

ZeroMQ Compatibility

Corresponds to ZMQ_LAST_ENDPOINT (32) option.

pub fn has_more(&self) -> bool

Check if the last received message has more frames coming.

Returns true if there are more frames in the current multipart message.

ZeroMQ Compatibility

Corresponds to ZMQ_RCVMORE (13) option.

pub fn events(&self) -> u32

Get the event state of the socket.

Returns a bitmask indicating ready-to-receive and ready-to-send states.

Returns

• 1 (POLLIN) - Socket is ready to receive
• 2 (POLLOUT) - Socket is ready to send
• 3 (POLLIN | POLLOUT) - Socket is ready for both

ZeroMQ Compatibility

Corresponds to ZMQ_EVENTS (15) option.

impl PairSocket

pub async fn bind( addr: impl ToSocketAddrsAsync, ) -> Result<(TcpListener, PairSocket), Error>

Bind to an address and accept the first connection.

PAIR sockets form an exclusive pair with exactly one peer.

Returns

A tuple of (listener, socket) where:

• listener can be used to accept additional connections if needed
• socket is ready to send/receive with the first peer

Example

use monocoque_zmtp::pair::PairSocket;

let (listener, mut socket) = PairSocket::bind("127.0.0.1:5555").await?;

pub async fn connect(addr: impl ToSocketAddrsAsync) -> Result<PairSocket, Error>

Connect to a remote PAIR socket, storing the endpoint for automatic reconnection.

Example

use monocoque_zmtp::pair::PairSocket;

let mut socket = PairSocket::connect("127.0.0.1:5555").await?;

pub async fn connect_with_options( addr: impl ToSocketAddrsAsync, options: SocketOptions, ) -> Result<PairSocket, Error>

Connect with custom options, storing the endpoint for reconnection.

pub async fn from_tcp(stream: TcpStream) -> Result<PairSocket, Error>

Create a new PAIR socket from a TCP stream with TCP_NODELAY enabled.

pub async fn from_tcp_with_options( stream: TcpStream, options: SocketOptions, ) -> Result<PairSocket, Error>

Create a new PAIR socket from a TCP stream with TCP_NODELAY and custom options.

pub fn is_connected(&self) -> bool

Check if the socket is currently connected.

pub async fn try_reconnect(&mut self) -> Result<(), Error>

Try to reconnect to the stored endpoint.

pub async fn recv_with_reconnect(&mut self) -> Result<Option<Vec<Bytes>>, Error>

Receive a message with automatic reconnection on EOF or network error.

If the socket was created with connect() and stores an endpoint, this method loops: on EOF or broken-pipe it clears the stream and calls try_reconnect() (which applies exponential backoff), then retries recv().

Respects max_reconnect_attempts - returns NotConnected when exhausted.

pub async fn send_with_reconnect( &mut self, msg: Vec<Bytes>, ) -> Result<(), Error>

Send a message with automatic reconnection on network error.

On BrokenPipe / ConnectionReset, write_from_buf() already sets stream = None, so the next loop iteration reconnects automatically.

Respects max_reconnect_attempts - returns NotConnected when exhausted.

impl PairSocket<InprocStream>

pub fn bind_inproc(endpoint: &str) -> Result<PairSocket<InprocStream>, Error>

Bind to an inproc endpoint.

Creates a new inproc endpoint that other sockets can connect to. Inproc endpoints must be bound before they can be connected to.

Arguments

• endpoint - Inproc URI (e.g., “inproc://my-endpoint”)

Example

use monocoque_zmtp::pair::PairSocket;

let socket = PairSocket::bind_inproc("inproc://my-pair")?;

pub fn bind_inproc_with_options( endpoint: &str, options: SocketOptions, ) -> Result<PairSocket<InprocStream>, Error>

Bind to an inproc endpoint with custom configuration and options.

pub fn connect_inproc(endpoint: &str) -> Result<PairSocket<InprocStream>, Error>

Connect to an inproc endpoint.

Connects to a previously bound inproc endpoint.

Arguments

• endpoint - Inproc URI (e.g., “inproc://my-endpoint”)

Example

use monocoque_zmtp::pair::PairSocket;

let socket = PairSocket::connect_inproc("inproc://my-pair")?;

pub fn connect_inproc_with_options( endpoint: &str, options: SocketOptions, ) -> Result<PairSocket<InprocStream>, Error>

Connect to an inproc endpoint with custom configuration and options.

Trait Implementations

impl ProxySocket for PairSocket

fn recv_multipart<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<Bytes>>, Error>> + 'async_trait>>

where 'life0: 'async_trait, PairSocket: 'async_trait,

Receive a multipart message from the socket.

fn send_multipart<'life0, 'async_trait>( &'life0 mut self, msg: Vec<Bytes>, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + 'async_trait>>

where 'life0: 'async_trait, PairSocket: 'async_trait,

Send a multipart message to the socket.

fn socket_desc(&self) -> &'static str

Get a description of the socket for logging.

impl<S> Socket for PairSocket<S>

where S: AsyncRead + AsyncWrite + Unpin + 'static,

fn send<'life0, 'async_trait>( &'life0 mut self, msg: Vec<Bytes>, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + 'async_trait>>

where 'life0: 'async_trait, PairSocket<S>: 'async_trait,

Send a multipart message on the socket.

fn recv<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<Bytes>>, Error>> + 'async_trait>>

where 'life0: 'async_trait, PairSocket<S>: 'async_trait,

Receive a multipart message from the socket.

fn socket_type(&self) -> SocketType

Get the socket type.

fn has_more(&self) -> bool

Check if socket has more message frames pending.