ZmqQueue

ZmqQueue is PUSH/PULL: competing consumers, each message reaches one of them, in turn and without a key.

A pipeline has a ventilator that pushes work, workers that pull it, and a sink that collects the results. Workers that dial one ventilator compete for its stream: each message reaches one of them, and none reaches two. PUSH hands the messages out in turn across the attached peers, whatever a message holds. Order holds per socket pair and nowhere else.

A subscription that binds its endpoint takes every message a peer pushes into it, whatever the name in frame 0 says. So a second kind of work takes an endpoint of its own.

The policy is ZmqQueuePublish, offered as Publish by queue::prelude. A publish to an endpoint that a subscription of this service binds reaches that subscription and nothing else. A message under the subscription's own name goes through: a retry copy, or a job the service feeds itself. A message under any other name is refused with ZmqError::Send, naming it and the subscription, because the subscription would receive it as its next job. So results and replies leave through a queue on an endpoint of its own, as the replies page shows.

A retry copy is a message this service publishes. On a bound queue the copy goes back into the queue, and whichever worker is free takes it. A queue that dials reads from a ventilator that takes nothing back, so its mount site names where the copies go; out_retry covers both forms.

Pipelines with PUSH/PULL builds a ventilator, competing workers and a sink on it.

Example

Two workers dial one ventilator, and each job it pushes reaches one of them. Their retry copies go to a queue this service binds, where the same handler takes them.

Macros
Manual

examples/zmq_broker_queue.rs
#[ruststream::app]
fn app() -> impl App {
    // The ventilator only pushes, so the retry copies go to a queue this service binds.
    let retries = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5560")).bindable();
    let first = retries.bind(Publish);
    let second = retries.bind(Publish);
    RustStream::new(AppInfo::new("workers", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::connect("tcp://ventilator:5555")),
            |b| {
                // Two workers dial the ventilator, and each job it pushes reaches one of them.
                b.include(work).out_retry(first).to("jobs");
                b.include(work).out_retry(second).to("jobs");
            },
        )
        .with_broker(retries, |b| {
            b.include(work);
        })
}

examples/manual/zmq_broker_queue.rs
fn app() -> impl App {
    // The ventilator only pushes, so the retry copies go to a queue this service binds.
    let retries = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5560")).bindable();
    let first = retries.bind(Publish);
    let second = retries.bind(Publish);
    RustStream::new(AppInfo::new("workers", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::connect("tcp://ventilator:5555")),
            |b| {
                // Two workers dial the ventilator, and each job it pushes reaches one of them.
                b.include(subscriber("jobs", Work).build())
                    .out_retry(first)
                    .to("jobs");
                b.include(subscriber("jobs", Work).build())
                    .out_retry(second)
                    .to("jobs");
            },
        )
        .with_broker(retries, |b| {
            b.include(subscriber("jobs", Work).build());
        })
}

Example​

Example