Struct ZmqSubscriber 

pub struct ZmqSubscriber { /* private fields */ }

A subscription on one of the one-way patterns - PUSH/PULL or PUB/SUB - yielding ZmqMessages singly or in batches.

A receive on a socket yields one multipart message, so a batch is collected on this side: it closes once it holds the size the mount named, or 20 ms after its first delivery, whichever comes first. The 20 ms is a constant of this crate.

Examples

A handler that takes a slice receives a whole batch, and the mount site caps its size:

use ruststream_zeromq::queue::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Job {
    id: u64,
}

#[subscriber("jobs")]
async fn drain(jobs: &[Job]) -> HandlerOutcome {
    tracing::info!(size = jobs.len(), "working on a batch");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
        |b| {
            b.include(drain.batch(nonzero!(32)));
        },
    )
}

Trait Implementations

impl BatchSubscriber for ZmqSubscriber

The transport has no batches of its own - a receive yields one multipart message - so they are assembled on the client, to the size the registration named. The deadline that closes a partial batch is the crate’s own (20 ms); the size is not, it arrives per subscription.

type Batch = Vec<ZmqMessage>

Container yielded by batches. Implementations choose between Vec, custom iterators, or anything else that yields the underlying Subscriber::Message.

fn batches( &mut self, size: NonZeroUsize, ) -> impl Stream<Item = Result<Self::Batch, ZmqError>> + Send + '_

Returns a stream of batches of at most size messages.

impl Debug for ZmqSubscriber

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

Formats the value using the given formatter.

impl Subscriber for ZmqSubscriber

type Message = ZmqMessage

The message type yielded by this subscriber.

type Error = ZmqError

The error type yielded by the stream when delivery fails.

fn stream( &mut self, ) -> impl Stream<Item = Result<ZmqMessage, ZmqError>> + Send + '_

Returns a stream of broker deliveries.