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

One of several workers that dial the same dispatcher and compete for its jobs. The dispatcher takes nothing back, so the retry copies go to a queue the worker binds itself:

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://dispatcher:5555")), |b| {
            b.include(work).out_retry(copies).to("jobs");
        })
        .with_broker(retries, |b| {
            b.include(work);
        })
}

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 {
    println!("working on job {}", job.id);
    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 {
    println!("rendering {}", job.scene);
    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.