Struct SubSocket 

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

A SUB socket for receiving filtered messages.

SUB sockets connect to PUB peers and filter messages by topic prefix. They’re used for:

• Event subscriptions
• Topic-based message filtering
• Many-to-one aggregation

ZeroMQ Compatibility

Compatible with zmq::SUB and zmq::PUB sockets from libzmq.

Example

use monocoque::zmq::SubSocket;

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

// Subscribe to topic
socket.subscribe(b"topic");

// Receive filtered messages
loop {
    match socket.recv().await? {
        Some(msg) => println!("Received: {:?}", msg),
        None => break, // Connection closed
    }
}

Implementations

impl SubSocket

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

Connect to a PUB peer and create a SUB socket.

Accepts TCP endpoints or raw socket addresses:

• "tcp://127.0.0.1:5555"
• "127.0.0.1:5555"

For IPC (Unix domain sockets), use SubSocket::connect_ipc().

Example

use monocoque::zmq::SubSocket;

let mut socket = SubSocket::connect("127.0.0.1:5555").await?;
socket.subscribe(b""); // Subscribe to all messages

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

Connect to a publisher with custom socket options.

Counterpart to connect for callers that need to set buffer sizes, timeouts, or subscriptions before the connection is made.

Errors

Returns an error if the address cannot be parsed or the connection and ZMTP handshake fail.

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, re-sending all active subscriptions.

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_ipc(path: &str) -> Result<SubSocket<UnixStream>>

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

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

• "ipc:///tmp/socket.sock"
• "/tmp/socket.sock"

Example

use monocoque::zmq::SubSocket;

let mut socket = SubSocket::connect_ipc("/tmp/pubsub.sock").await?;
socket.subscribe(b"");

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

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

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

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

Example

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

let stream = TcpStream::connect("127.0.0.1:5555").await?;
let socket = SubSocket::from_tcp_with_options(
    stream,
    SocketOptions::default()
        .with_recv_hwm(500)
        .with_buffer_sizes(4096, 4096)
).await?;

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

where Stream: AsyncRead + AsyncWrite + Unpin,

Create a SUB socket from any stream with custom options.

impl<S> SubSocket<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 subscribe(&mut self, topic: &[u8]) -> Result<()>

Subscribe to messages matching the given topic prefix.

Empty topic subscribes to all messages.

This sends a subscription message to the PUB socket.

pub async fn unsubscribe(&mut self, topic: &[u8]) -> Result<()>

Unsubscribe from messages matching the given topic prefix.

This sends an unsubscription message to the PUB socket.

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

Receive a multipart message.

Only messages matching subscribed topics will be received. Returns None if the connection is closed.

pub async fn recv_into(&mut self, out: &mut Vec<Bytes>) -> Result<bool>

Receive a matching message into a caller-provided buffer, reusing its allocation. Allocation-free counterpart to recv: filtered messages are dropped without allocating. Returns Ok(true) on a matching message, Ok(false) on EOF.

pub fn try_recv_into(&mut self, out: &mut Vec<Bytes>) -> Result<bool>

Try to receive a matching message into out without a kernel read.

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 fn options_mut(&mut self) -> &mut SocketOptions

Get a mutable reference to this socket’s options.

impl SubSocket<UnixStream>

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

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

This method provides full control over socket behavior through SocketOptions.