Struct RepSocket 

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

A REP socket for synchronous reply patterns.

REP sockets enforce strict alternation between receive and send:

• Must call recv() to get a request
• Must call send() to reply before next recv()
• Automatically handles routing envelopes

They’re used for:

• Synchronous RPC servers
• Request-reply protocols
• Service endpoints

ZeroMQ Compatibility

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

Example

use monocoque::zmq::RepSocket;
use monocoque_core::rt::TcpListener;
use bytes::Bytes;

// Bind and accept
let listener = TcpListener::bind("127.0.0.1:5555").await?;
let (stream, _) = listener.accept().await?;
let mut socket = RepSocket::from_tcp(stream).await?;

loop {
    // Receive request
    if let Ok(Some(request)) = socket.recv().await {
        println!("Got request: {:?}", request);
         
        // Send reply
        socket.send(vec![Bytes::from("REPLY")]).await?;
    }
}

Implementations

impl RepSocket

pub async fn bind(addr: impl ToSocketAddrs) -> Result<(TcpListener, Self)>

Bind to addr, accept one connection, and return a ready REP socket.

Returns the TcpListener so the caller can accept further connections.

Example

use monocoque::zmq::RepSocket;
use bytes::Bytes;

let (_listener, mut socket) = RepSocket::bind("127.0.0.1:5555").await?;
if let Ok(Some(req)) = socket.recv().await {
    socket.send(vec![Bytes::from("PONG")]).await?;
}

pub async fn bind_with_options( addr: impl ToSocketAddrs, options: SocketOptions, ) -> Result<(TcpListener, Self)>

Bind with custom socket options.

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

Create a REP socket from an existing TCP stream.

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

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

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

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

where Stream: AsyncRead + AsyncWrite + Unpin,

Create a REP socket from any stream with custom options.

impl<S> RepSocket<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 recv(&mut self) -> Result<Option<Vec<Bytes>>>

Receive a request message.

This blocks until a request is received. The routing envelope is automatically extracted and stored for the subsequent send() call.

Returns

• Some(msg) - Received a request (content only, envelope stripped)
• None - Connection closed gracefully or error occurred

Example

use monocoque::zmq::RepSocket;

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

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. \n /// # 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 send(&mut self, msg: Vec<Bytes>) -> Result<()>

Send a reply message.

This must be called after recv() and automatically uses the stored routing envelope from the request.

Arguments

• msg - Vector of message frames (parts) to send as reply

Errors

Returns an error if:

• Called without first calling recv()
• The underlying connection is closed

Example

use monocoque::zmq::RepSocket;
use bytes::Bytes;

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

// Send multi-part reply
socket.send(vec![
    Bytes::from("Status: OK"),
    Bytes::from("Data: ..."),
]).await?;

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

Get a mutable reference to this socket’s options.

impl RepSocket<UnixStream>

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

Create a REP 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 REP socket from an existing Unix stream with custom options.

This method provides full control over socket behavior through SocketOptions.