Struct XPubSocket 

pub struct XPubSocket { /* private fields */ }

XPUB (Extended Publisher) socket.

Receives subscription events and broadcasts messages to matching subscribers.

Features

• Subscription tracking: Know what topics subscribers want
• Verbose mode: Report all subscriptions (including duplicates)
• Manual mode: Explicit subscription control
• Welcome messages: Send initial message to new subscribers

Examples

use monocoque_zmtp::xpub::XPubSocket;
use bytes::Bytes;

let mut xpub = XPubSocket::bind("127.0.0.1:5555").await?;
     
    loop {
        // Receive subscription events from subscribers
        if let Some(event) = xpub.recv_subscription().await? {
            println!("Subscription event: {:?}", event);
        }
         
        // Broadcast messages to matching subscribers
        xpub.send(vec![Bytes::from("topic"), Bytes::from("data")]).await?;
    }

Implementations

impl XPubSocket

pub async fn bind(addr: &str) -> Result<XPubSocket, Error>

Bind to an address and start listening for subscribers.

Examples

let xpub = XPubSocket::bind("127.0.0.1:5555").await?;

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

Bind with custom socket options.

Honors options.reuse_port: when set, the listener is bound with SO_REUSEPORT so several XPUB acceptors can share one port.

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

Accept new subscriber connections (non-blocking).

Call this periodically to accept new subscribers.

pub async fn recv_subscription( &mut self, ) -> Result<Option<SubscriptionEvent>, Error>

Receive a subscription event from subscribers (non-blocking).

Returns None if no events are available.

Examples

if let Some(event) = xpub.recv_subscription().await? {
    match event {
        monocoque_core::subscription::SubscriptionEvent::Subscribe(topic) => {
            println!("New subscription: {:?}", topic);
        }
        monocoque_core::subscription::SubscriptionEvent::Unsubscribe(topic) => {
            println!("Unsubscription: {:?}", topic);
        }
    }
}

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

Broadcast a message to all matching subscribers.

Only subscribers whose subscriptions match the message’s first frame will receive it.

Examples

xpub.send(vec![
    Bytes::from("topic.temperature"),
    Bytes::from("23.5"),
]).await?;

pub fn subscriber_count(&self) -> usize

Get the number of active subscribers.

pub fn local_addr(&self) -> Result<SocketAddr, Error>

Get the local address.

pub const fn socket_type(&self) -> SocketType

Get the socket type.

pub fn has_more(&self) -> bool

Check if the last received message has more frames coming.

For XPUB, subscription events are always single-frame.

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 (has pending subscription events)
• 2 (POLLOUT) - Socket is ready to send (has active subscribers)
• 3 (POLLIN | POLLOUT) - Socket is ready for both

ZeroMQ Compatibility

Corresponds to ZMQ_EVENTS (15) option.

pub fn set_verbose(&mut self, verbose: bool)

Set verbose mode.

When enabled, all subscription messages are reported (including duplicates).

pub fn set_manual(&mut self, manual: bool)

Set manual mode.

When enabled, subscriptions must be explicitly approved by calling send_subscription().

pub async fn connect_upstream(&mut self, addr: &str) -> Result<(), Error>

Connect to an upstream publisher so that subscription events can be forwarded.

The upstream is typically a PUB or XSUB socket. After calling this method, send_subscription() (manual mode) writes subscription messages to the upstream connection, causing the upstream publisher to start or stop delivering matching messages.

Examples

let mut xpub = XPubSocket::bind("127.0.0.1:5556").await?;
xpub.set_manual(true);
xpub.connect_upstream("127.0.0.1:5555").await?;

// Receive a subscription from a downstream client and forward it upstream.
if let Some(event) = xpub.recv_subscription().await? {
    xpub.send_subscription(event).await?;
}

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

Manually send a subscription event to the upstream connection.

Requires both manual mode (set_manual(true)) and an upstream connection (connect_upstream()). Writes the subscription message directly to the upstream publisher so it starts (or stops) delivering matching messages.

Trait Implementations

impl Debug for XPubSocket

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl ProxySocket for XPubSocket

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, XPubSocket: '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, XPubSocket: 'async_trait,

Send a multipart message to the socket.

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

Get a description of the socket for logging.

impl Socket for XPubSocket

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

where 'life0: 'async_trait, XPubSocket: '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, XPubSocket: '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.