Struct XSubSocket 

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

XSUB (Extended Subscriber) socket.

Receives data messages and can send subscription messages upstream.

Features

• Dynamic subscriptions: Subscribe/unsubscribe at runtime
• Subscription forwarding: Forward subscriptions in proxies
• Verbose unsubscribe: Optionally send explicit unsubscribe messages

Examples

use monocoque_zmtp::xsub::XSubSocket;
use bytes::Bytes;

let mut xsub = XSubSocket::connect("127.0.0.1:5555").await?;
     
    // Subscribe to topics
    xsub.subscribe("topic.").await?;

    // Receive messages
    if let Some(msg) = xsub.recv().await? {
        println!("Received: {:?}", msg);
    }

    // Unsubscribe
xsub.unsubscribe("topic.").await?;

Implementations

impl<S> XSubSocket<S>

where S: AsyncRead + AsyncWrite + Unpin,

pub async fn new(stream: S) -> Result<XSubSocket<S>, Error>

Create a new XSUB socket from a stream.

pub async fn with_options( stream: S, options: SocketOptions, ) -> Result<XSubSocket<S>, Error>

Create a new XSUB socket with custom configuration and options.

pub async fn subscribe(&mut self, prefix: impl Into<Bytes>) -> Result<(), Error>

Subscribe to messages with the given prefix.

Sends a subscription message upstream to the publisher.

Examples

// Subscribe to all messages starting with "topic."
xsub.subscribe("topic.").await?;

// Subscribe to all messages (empty prefix)
xsub.subscribe("").await?;

pub async fn unsubscribe( &mut self, prefix: impl Into<Bytes>, ) -> Result<(), Error>

Unsubscribe from messages with the given prefix.

Optionally sends an unsubscribe message upstream (if verbose mode enabled).

Examples

let prefix = Bytes::from_static(b"topic.");
xsub.unsubscribe(prefix).await?;

pub async fn send_subscription_event( &mut self, event: SubscriptionEvent, ) -> Result<(), Error>

Send a raw subscription event upstream (for proxies).

This allows forwarding subscription messages in broker patterns.

Examples

let event = SubscriptionEvent::Subscribe(Bytes::from("topic"));
xsub.send_subscription_event(event).await?;

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

Receive a data message (non-blocking).

Returns None if no message is available.

Examples

if let Some(msg) = xsub.recv().await? {
    for frame in msg {
        println!("Frame: {:?}", frame);
    }
}

pub fn subscription_count(&self) -> usize

Get the number of active subscriptions.

pub fn is_subscribed(&self, topic: &[u8]) -> bool

Check if subscribed to a specific topic.

pub fn subscriptions(&self) -> Vec<Subscription>

Get all subscriptions.

pub const fn socket_type(&self) -> SocketType

Get the socket type.

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.

impl XSubSocket

pub async fn connect(addr: &str) -> Result<XSubSocket, Error>

Connect to a publisher, storing the endpoint for automatic reconnection.

Examples

let xsub = XSubSocket::connect("127.0.0.1:5555").await?;

pub async fn connect_with_options( addr: &str, options: SocketOptions, ) -> Result<XSubSocket, Error>

Connect with custom socket options, storing the endpoint for automatic reconnection.

pub fn is_connected(&self) -> bool

Check if the socket is currently connected.

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

Try to reconnect to the stored endpoint, re-sending all active subscriptions.

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

Receive a message with automatic reconnection on EOF or network error.

Respects max_reconnect_attempts - returns NotConnected when exhausted.

Trait Implementations

impl ProxySocket for XSubSocket

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

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

where 'life0: 'async_trait, XSubSocket: 'async_trait,

Send a multipart message to the socket.

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

Get a description of the socket for logging.

impl<S> Socket for XSubSocket<S>

where S: AsyncRead + AsyncWrite + Unpin + 'static,

fn send<'life0, 'async_trait>( &'life0 mut self, msg: Vec<Bytes>, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + 'async_trait>>

where 'life0: 'async_trait, XSubSocket<S>: 'async_trait,

Send a multipart message on the socket.

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

where 'life0: 'async_trait, XSubSocket<S>: 'async_trait,

Receive a multipart message from the socket.

fn socket_type(&self) -> SocketType

Get the socket type.

fn has_more(&self) -> bool

Check if socket has more message frames pending.