Skip to main content

Pipelines with PUSH/PULL

This guide builds the pipeline of the ZeroMQ guide: a ventilator pushes jobs, workers compete for them, and a sink collects the results.

The example has two services. The worker takes jobs from the ventilator and pushes a result for each to the sink. The sink collects the results in batches. The ventilator is any peer that pushes jobs, such as the Python peer of Getting started.

Who binds​

The stable side of a connection binds, and the side that comes and goes dials. In a pipeline the ventilator and the sink are stable, so they bind. The workers dial both, and one more worker process is one more dial.

ZmqEndpoint::bind and ZmqEndpoint::connect state the side. The side is part of the broker's type: a broker built on connect is a ZmqQueue<Connect>. It decides what the broker may publish and where a retry copy can go, so a mistake is caught at compile time or at startup, not in production.

The worker​

src/main.rs
use ruststream_zeromq::queue::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, PartialEq, Deserialize)]
struct Job {
    id: u64,
    items: u32,
}

/// What the sink collects: one result per job.
#[derive(Debug, PartialEq, Serialize, Outgoing)]
#[outgoing(name = "results")]
struct Done {
    id: u64,
    items: u32,
}

#[subscriber("jobs", reply)]
async fn work(job: &Job) -> Done {
    tracing::info!(job.id, job.items, "job worked");
    Done {
        id: job.id,
        items: job.items,
    }
}

work takes a job and returns a Done. Done names its destination itself, results, so the subscriber clause carries a bare reply.

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    // The sink listens; every worker dials it and only pushes there.
    let sink = ZmqQueue::new(ZmqEndpoint::connect("tcp://sink:5558")).bindable();
    // The retry copies come back through a queue this worker binds for itself.
    let retries = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5559")).bindable();
    let (results, retried_results) = (sink.bind(Publish), sink.bind(Publish));
    let copies = retries.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker_labeled(
            "ventilator",
            ZmqQueue::new(ZmqEndpoint::connect("tcp://ventilator:5557")),
            |b| {
                b.include(work)
                    .out_reply(results)
                    .out_retry(copies)
                    .to("jobs");
            },
        )
        .with_broker_labeled("retries", retries, |b| {
            b.include(work).out_reply(retried_results);
        })
        .with_broker_labeled("sink", sink, |_b| {})
}

The ventilator broker dials the ventilator, and its subscription reads the jobs. Workers that dial one ventilator compete: PUSH hands each job to one worker, in turn.

The results leave through a second broker, sink, which dials the sink. bindable() turns a broker into one that publish policies can be bound to, and sink.bind(Publish) is a policy that publishes through it. .out_reply(results) sends every Done there. The worker cannot reply through the ventilator broker: the ventilator only sends, so the broker refuses every publish once its subscription has dialed. Replies covers the refusals.

A subscription that dials also names where its retry copies go, even when its handler never asks for a retry. Here they go to retries, a queue the worker binds for itself, and the same handler reads it. A copy carries the name jobs, the name of the subscription that takes it back. Retries and dead letters covers retries in full.

The sink​

src/main.rs
use ruststream_zeromq::queue::prelude::*;
use serde::Deserialize;

#[derive(Debug, PartialEq, Deserialize)]
struct Done {
    id: u64,
    items: u32,
}

#[subscriber("results")]
async fn collect(results: &[Done]) -> HandlerOutcome {
    let items: u32 = results.iter().map(|done| done.items).sum();
    tracing::info!(results = results.len(), items, "results collected");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("sink", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5558")),
        |b| {
            // At most 32 results per call.
            b.include(collect.batch(nonzero!(32)));
        },
    )
}

The sink binds the address the workers dial. A bound PULL socket takes whatever any worker pushes to it, so one subscription reads the results of every worker. collect takes a slice, so it receives batches. A batch closes once it holds 32 results, or 20 ms after its first result, whichever comes first. The frames on the wire do not change: the batch is gathered on the sink's side.

One kind of work per endpoint​

PUSH and PULL do not look at the name frame. A bound queue hands its subscription every message pushed to it, whatever the name says. So a second kind of work, such as reports next to jobs, gets an endpoint of its own and a broker of its own. A second subscription on a bound endpoint is refused at startup, and the error names both subscriptions.

Order and keys​

PUSH hands messages to the attached workers in turn and does not read a key. Two jobs of one customer may run on two workers at the same time. A handler that must see one customer's jobs in order runs in one worker process. Inside one process, workers at the mount site runs several deliveries of a handler at once. A ZeroMQ message carries no key, so lanes keyed with by_key take its deliveries in turn.

Testing the pipeline​

Each example file holds a test that starts its service with TestApp::start. The worker's test pushes a job as the ventilator would and reads the result published to results on the sink broker. The sink's test pushes three results at once and checks that collect received them as one batch. Testing a ZeroMQ service covers the harness.