Struct ZmqQueue 

pub struct ZmqQueue<Role = Bind> { /* private fields */ }

The PUSH/PULL queue: each message reaches one of the competing consumers.

Role is the side of the endpoint this process takes, fixed by the endpoint’s constructor: Bind for ZmqEndpoint::bind, Connect for ZmqEndpoint::connect. It decides where a retry copy goes. A subscription that binds takes a copy published to its own listener, so a mount site names no destination; one that dials reads from a PUSH peer that takes nothing, so a mount site names where its copies go.

Examples

A worker that binds its queue retries on it. The copy goes back into the queue, whichever worker is free takes it, and the framework’s retry-count header caps the deliveries at five:

use std::time::Duration;

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

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

#[subscriber("jobs")]
async fn handle(job: &Job) -> HandlerOutcome {
    if job.id == 0 {
        return HandlerOutcome::retry_after(Duration::from_secs(30));
    }
    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(handle)
                .max_attempts(nonzero!(5u32))
                .out_retry(Publish);
        },
    )
}

One of several workers that dial the same ventilator and compete for its jobs. The ventilator takes nothing back, so the retry copies go to a queue the worker binds itself, where the same handler takes them. The copies carry the framework’s counter, so the cap holds across both subscriptions:

use std::time::Duration;

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

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

#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    if job.id == 0 {
        return HandlerOutcome::retry_after(Duration::from_secs(30));
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let retries = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5560")).bindable();
    let copies = retries.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(ZmqEndpoint::connect("tcp://ventilator:5555")), |b| {
            b.include(work)
                .max_attempts(nonzero!(5u32))
                .out_retry(copies)
                .to("jobs");
        })
        .with_broker(retries, |b| {
            b.include(work).max_attempts(nonzero!(5u32));
        })
}

Implementations

impl<Role: EndpointRole> ZmqQueue<Role>

pub fn new(endpoint: ZmqEndpoint<Role>) -> Self

Records the endpoint. No I/O.

Examples

A worker that listens for the jobs its producers push:

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

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

#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "working on the job");
    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(work);
        },
    )
}

impl<Role> ZmqQueue<Role>

pub const fn read_ahead(self, read_ahead: NonZeroUsize) -> Self

How many deliveries a subscription reads off its socket ahead of the handler; 1000 unless set, the receive high-water mark ZeroMQ itself gives a socket.

Past this bound the subscription stops reading, the socket’s buffers fill, and the sender waits: a handler slower than the wire slows the sender down rather than growing this process’s memory.

Examples

A worker whose renders take seconds holds 64 jobs ahead, and the producers wait beyond that:

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

#[derive(Deserialize)]
struct Render {
    scene: String,
}

#[subscriber("renders")]
async fn render(job: &Render) -> HandlerOutcome {
    tracing::info!(scene = %job.scene, "rendering");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let renders = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555"))
        .read_ahead(nonzero!(64_usize));
    RustStream::new(AppInfo::new("renderer", "0.1.0")).with_broker(renders, |b| {
        b.include(render);
    })
}

pub fn publisher(&self) -> ZmqQueuePublisher

A publisher sharing this queue’s state; buildable before connect.

Trait Implementations

impl<Role: EndpointRole> Broker for ZmqQueue<Role>

type Error = ZmqError

The error type returned by broker-level operations.

type Connected = ConnectedZmqQueue<Role>

The connected form of this broker: the typed witness that connect succeeded.

async fn connect(self) -> Result<Self::Connected, Self::Error>

Establishes the connection to the broker, consuming the unconnected form.

fn bindable(self) -> Bindable<Self>

where Self: 'static,

Wraps the broker for publisher-token minting before registration: the returned Bindable hands out Bound tokens via its bind, and is itself what with_broker takes.

impl<Role> Clone for ZmqQueue<Role>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<Role> Debug for ZmqQueue<Role>

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

Formats the value using the given formatter.

impl<Role: EndpointRole> DescribeServer for ZmqQueue<Role>

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl<Role: EndpointRole> InProcess for ZmqQueue<Role>

async fn connect_in_process(self) -> Result<Self::Connected, Self::Error>

Connects to the in-process transport, consuming the configuration.