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 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 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.