Struct DealerSocket 

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

A DEALER socket for asynchronous request-reply patterns.

DEALER sockets are fair-queuing clients that distribute messages across multiple server endpoints. They’re used for:

• Load-balanced request-reply
• Async RPC clients
• Worker pools

ZeroMQ Compatibility

Compatible with zmq::DEALER and zmq::ROUTER sockets from libzmq.

Example

use monocoque::zmq::DealerSocket;
use bytes::Bytes;

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

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

// Receive reply
if let Ok(Some(reply)) = socket.recv().await {
    println!("Got reply: {:?}", reply);
}

Implementations

impl DealerSocket

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

Connect to a ZeroMQ peer and create a DEALER 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 fails (network unreachable, connection refused, etc.)
• DNS resolution fails for TCP endpoints
• Invalid endpoint format

Example

use monocoque::zmq::DealerSocket;

// TCP connection
let socket1 = DealerSocket::connect("tcp://127.0.0.1:5555").await?;

// IPC connection (Unix only)
#[cfg(unix)]
let socket2 = DealerSocket::connect("ipc:///tmp/socket.sock").await?;

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

Connect to a ZeroMQ peer with custom socket options.

Stores the endpoint so the socket can reconnect automatically after failures.

Example

use monocoque::zmq::{DealerSocket, SocketOptions};
use std::time::Duration;

let mut socket = DealerSocket::connect_with_options(
    "tcp://127.0.0.1:5555",
    SocketOptions::default().with_send_hwm(100),
).await?;

pub async fn connect_ipc(path: &str) -> Result<DealerSocket<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::DealerSocket;

let mut socket = DealerSocket::connect_ipc("/tmp/dealer.sock").await?;

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

Bind to an address and accept the first connection.

This creates a server-side DEALER socket that accepts incoming connections. Useful for broker patterns where workers (REP sockets) connect to a DEALER backend.

Returns

A tuple of (listener, socket) where:

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

Errors

Returns an error if:

• The address is already in use
• Permission denied (e.g., binding to privileged port without root)
• Invalid address format

Example

use monocoque::zmq::DealerSocket;

// Bind DEALER backend for worker connections
let (listener, socket) = DealerSocket::bind("127.0.0.1:5556").await?;

// Use socket for first connection
// Accept more connections from listener if needed:
// let (stream, _) = listener.accept().await?;
// let socket2 = DealerSocket::from_tcp(stream).await?;

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

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

This method automatically enables TCP_NODELAY for optimal performance, preventing Nagle’s algorithm from buffering small packets.

Uses default buffer sizes (8KB) and socket options. For custom configuration, use with_options().

Example

use monocoque::zmq::DealerSocket;
use monocoque_core::rt::TcpStream;

let stream = TcpStream::connect("127.0.0.1:5555").await?;
let socket = DealerSocket::from_tcp(stream).await?;

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

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

Provides full control over buffer sizes, HWM, timeouts, etc. through SocketOptions.

Examples

use monocoque::zmq::{DealerSocket, SocketOptions};
use monocoque_core::rt::TcpStream;

let stream = TcpStream::connect("127.0.0.1:5555").await?;

// Customize HWM only (uses default 8KB buffers)
let socket = DealerSocket::from_tcp_with_options(
    stream,
    SocketOptions::default().with_send_hwm(100)
).await?;

// Customize both buffers and HWM
let socket = DealerSocket::from_tcp_with_options(
    stream,
    SocketOptions::default()
        .with_buffer_sizes(4096, 4096)  // 4KB buffers for low latency
        .with_send_hwm(100)
).await?;

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

where Stream: AsyncRead + AsyncWrite + Unpin,

Create a DEALER socket from any stream with custom options.

This is the most flexible constructor - works with TCP, Unix, or in-memory streams. Useful for testing with duplex streams.

Example

use monocoque::zmq::{DealerSocket, SocketOptions};
use monocoque_core::rt::TcpStream;

let stream = TcpStream::connect("127.0.0.1:5555").await?;
let socket = DealerSocket::with_options(
    stream,
    SocketOptions::default().with_send_hwm(10)
).await?;

impl<S> DealerSocket<S>

where S: AsyncRead + AsyncWrite + Unpin,

pub fn monitor(&mut self) -> SocketMonitor

Enable monitoring for this socket.

Returns a receiver for socket lifecycle events. Once enabled, the socket will emit events like Connected, Disconnected, etc.

Example

use monocoque::zmq::{DealerSocket, SocketEvent};

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

// Spawn task to handle events
monocoque::rt::spawn_detached(async move {
    while let Ok(event) = monitor.recv_async().await {
        println!("Socket event: {}", event);
    }
});

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

Send a multipart message.

Messages are sent asynchronously - this returns immediately after queuing the message for transmission.

Errors

Returns an error if the underlying connection is closed or broken.

Example

socket.send(vec![
    Bytes::from("part1"),
    Bytes::from("part2"),
]).await?;

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

Send a message to the internal buffer without flushing.

Use this for batching multiple messages before a single flush. Call flush() to send all buffered messages.

Example

// Batch 100 messages
for i in 0..100 {
    socket.send_buffered(vec![Bytes::from(format!("msg {}", i))])?;
}
// Single I/O operation for all 100 messages
socket.flush().await?;

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

Flush all buffered messages to the network.

Sends all messages buffered by send_buffered() in a single I/O operation.

pub async fn send_batch(&mut self, messages: &[Vec<Bytes>]) -> Result<()>

Send multiple messages in a single batch (convenience method).

This is equivalent to calling send_buffered() for each message followed by flush(), but more ergonomic.

Example

let messages = vec![
    vec![Bytes::from("msg1")],
    vec![Bytes::from("msg2")],
    vec![Bytes::from("msg3")],
];
socket.send_batch(&messages).await?;

pub fn buffered_bytes(&self) -> usize

Get the number of bytes currently buffered.

pub fn events(&self) -> u32

Get the socket type.

ZeroMQ Compatibility

Get current socket events (read/write readiness).

Returns a bitmask:

• 1 (POLLIN): Can receive without blocking
• 2 (POLLOUT): Can send without blocking

ZeroMQ Compatibility

Corresponds to ZMQ_EVENTS (15) option.

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

Receive a multipart message.

Returns None if the connection is closed.

Example

while let Ok(Some(msg)) = socket.recv().await {
    println!("Received {} parts", msg.len());
}

impl<S> DealerSocket<S>

where S: AsyncRead + AsyncWrite + Unpin,

pub const fn socket_type(&self) -> SocketType

Get the socket type.

Always returns SocketType::Dealer for DEALER sockets.

ZeroMQ Compatibility

Corresponds to ZMQ_TYPE (16) socket option.

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

Get the last connected endpoint as a string.

Returns the endpoint this socket connected to, if any.

ZeroMQ Compatibility

Corresponds to ZMQ_LAST_ENDPOINT (32) socket option.

pub fn has_more(&self) -> bool

Check if more message frames are expected.

For multipart messages, this indicates if more frames follow.

ZeroMQ Compatibility

Corresponds to ZMQ_RCVMORE (13) socket option.

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

Get mutable access to socket options.

Allows runtime modification of socket behavior.

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

Get immutable access to socket options.

impl DealerSocket<UnixStream>

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

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

Trait Implementations

impl ProxySocket for DealerSocket<TcpStream>

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

where Self: 'async_trait, 'life0: '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<()>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Send a multipart message to the socket.

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

Get a description of the socket for logging.