Struct ReqSocket 

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

A REQ socket for synchronous request-reply patterns.

REQ sockets enforce strict alternation between send and receive:

• Must call send() before recv()
• Must call recv() before next send()

They’re used for:

• Synchronous RPC clients
• Request-reply protocols
• Client-server communication

ZeroMQ Compatibility

Compatible with zmq::REQ and zmq::REP sockets from libzmq.

Example

use monocoque::zmq::ReqSocket;
use bytes::Bytes;

// Connect to server
let mut socket = ReqSocket::connect("127.0.0.1:5555").await?;

// Send request
socket.send(vec![Bytes::from("REQUEST")]).await?;

// Must receive before next send
if let Ok(Some(reply)) = socket.recv().await {
    println!("Got reply: {:?}", reply);
}

// Now can send again
socket.send(vec![Bytes::from("ANOTHER")]).await?;
if let Ok(Some(reply)) = socket.recv().await {
    println!("Got reply: {:?}", reply);
}

Implementations

impl ReqSocket

pub async fn connect(endpoint: &str) -> Result<Self>

Connect to a ZeroMQ peer and create a REQ socket.

Supports both TCP and IPC endpoints:

• TCP: "tcp://127.0.0.1:5555" or "127.0.0.1:5555"
• IPC: "ipc:///tmp/socket.sock" (Unix only)

Arguments

• endpoint - Endpoint to connect to

Errors

Returns an error if the connection or handshake fails.

Example

use monocoque::zmq::ReqSocket;

let socket = ReqSocket::connect("tcp://127.0.0.1:5555").await?;

pub fn is_connected(&self) -> bool

Check if the socket is currently connected.

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

Try to reconnect to the stored endpoint.

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

Send with automatic reconnection on network error.

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

Receive with automatic reconnection on EOF or network error.

pub async fn connect_with_options( endpoint: &str, options: SocketOptions, ) -> Result<Self>

Connect to a ZeroMQ peer with custom socket options.

This allows configuring timeouts and other options before connection.

Example

use monocoque::zmq::ReqSocket;
use monocoque::SocketOptions;
use std::time::Duration;

let options = SocketOptions::default()
    .with_send_timeout(Duration::from_secs(5))
    .with_recv_timeout(Duration::from_secs(10));

let socket = ReqSocket::connect_with_options(
    "tcp://127.0.0.1:5555",
    options
).await?;

pub async fn connect_ipc(path: &str) -> Result<ReqSocket<UnixStream>>

Connect to a ZeroMQ peer via IPC (Unix domain sockets).

Unix-only. Accepts IPC paths with or without ipc:// prefix.

Example

use monocoque::zmq::ReqSocket;

let socket = ReqSocket::connect_ipc("/tmp/req.sock").await?;

pub async fn from_tcp(stream: TcpStream) -> Result<Self>

Create a REQ socket from a TCP stream with TCP_NODELAY enabled.

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

Create a REQ socket from a TCP stream with custom socket options.

pub async fn with_options<Stream>( stream: Stream, options: SocketOptions, ) -> Result<ReqSocket<Stream>>

where Stream: AsyncRead + AsyncWrite + Unpin,

Create a REQ socket from any stream with custom options.

impl<S> ReqSocket<S>

where S: AsyncRead + AsyncWrite + Unpin,

pub fn monitor(&mut self) -> SocketMonitor

Enable monitoring for this socket.

Returns a receiver for socket lifecycle events.

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

Send a multipart message.

This enforces the REQ state machine - you must call recv() before calling send() again.

Arguments

• msg - Vector of message frames (parts)

Errors

Returns an error if:

• Called while awaiting a reply (must call recv() first)
• The underlying connection is closed

Example

use monocoque::zmq::ReqSocket;
use bytes::Bytes;

// Send single-part message
socket.send(vec![Bytes::from("Hello")]).await?;

// Send multi-part message
socket.send(vec![
    Bytes::from("Part 1"),
    Bytes::from("Part 2"),
]).await?;

pub const fn socket_type() -> 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.

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

Receive a multipart message.

This blocks until a reply is received. You must call this after send() before calling send() again.

Returns

• Ok(Some(msg)) - Received a multipart message
• Ok(None) - Connection closed gracefully (peer disconnected)
• Err(e) - Network or I/O error

Example

use monocoque::zmq::ReqSocket;

if let Some(reply) = socket.recv().await? {
    for (i, frame) in reply.iter().enumerate() {
        println!("Frame {}: {:?}", i, frame);
    }
}

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

Get a reference to the socket options.

Example

use monocoque::zmq::ReqSocket;

let timeout = socket.options().recv_timeout;
println!("Receive timeout: {:?}", timeout);

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

Get a mutable reference to the socket options.

Example

use monocoque::zmq::ReqSocket;
use std::time::Duration;

socket.options_mut().recv_timeout = Some(Duration::from_secs(30));

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

Set the socket options.

Example

use monocoque::zmq::ReqSocket;
use monocoque::SocketOptions;
use std::time::Duration;

let options = SocketOptions::default()
    .with_send_timeout(Duration::from_secs(5))
    .with_recv_timeout(Duration::from_secs(10));

socket.set_options(options);

impl ReqSocket<UnixStream>

pub async fn from_unix_stream(stream: UnixStream) -> Result<Self>

Create a REQ socket from an existing Unix domain socket stream (IPC).

pub async fn from_unix_stream_with_options( stream: UnixStream, options: SocketOptions, ) -> Result<Self>

Create a REQ socket from an existing Unix stream with custom options.

This method provides full control over socket behavior through SocketOptions.