Skip to main content

Broadcast with PUB/SUB

This guide announces events on PUB/SUB: one service publishes, and every watcher that subscribed receives a copy.

The example has two services. The worker takes jobs and announces each finished job on a fan-out. The watcher subscribes to every job announcement.

The announcer​

src/main.rs
// One file, two patterns: the crate prelude names each policy after its pattern.
use ruststream_zeromq::prelude::*;
use serde::{Deserialize, Serialize};

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

/// The announcement, under a name every `jobs.` watcher matches.
#[derive(Debug, PartialEq, Serialize, Outgoing)]
#[outgoing(name = "jobs.done")]
struct JobDone {
    id: u64,
}

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

#[ruststream::app]
fn app() -> impl App {
    // The watchers dial this endpoint; the announcer is the stable side, so it binds.
    let events = ZmqFanout::new(ZmqEndpoint::bind("tcp://0.0.0.0:5556")).bindable();
    let announce = events.bind(ZmqFanoutPublish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                b.include(work).out_reply(announce);
            },
        )
        .with_broker(events, |_b| {})
}

The worker reads jobs on a ZmqQueue and announces on a ZmqFanout, the PUB/SUB pattern. A file that mounts two patterns imports the crate's prelude, which names each policy after its pattern: ZmqFanoutPublish here. The announcer is the stable side, so it binds, and the watchers dial it. .out_reply(announce) publishes every JobDone under its name, jobs.done.

The watcher​

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

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

/// Every job announcement: `jobs.done`, `jobs.failed`, and whatever `jobs.` name comes next.
#[subscriber("jobs.")]
async fn watch(event: &JobEvent, ctx: &mut Context) -> HandlerOutcome {
    tracing::info!(event.id, name = ctx.name(), "job announced");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // A subscription that dials names where its retry copies go: a queue this service binds.
    let retries = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5560")).bindable();
    let copies = retries.bind(ZmqQueuePublish);
    RustStream::new(AppInfo::new("watcher", "0.1.0"))
        .with_broker(
            ZmqFanout::new(ZmqEndpoint::connect("tcp://worker:5556")),
            |b| {
                b.include(watch).out_retry(copies).to("jobs.");
            },
        )
        .with_broker(retries, |b| {
            b.include(watch);
        })
}

On a fan-out the subscription name is a prefix of the message name, compared byte by byte. jobs. receives jobs.done, jobs.failed and any other jobs. name, and nothing else. ctx.name() is the name the message arrived under. Every watcher that dials the announcer receives its own copy of each announcement.

The watcher dials, so its mount names where retry copies go, even though watch never asks for a retry. The copies go to a queue the watcher binds for itself, and the same handler reads it. Retries and dead letters covers this rule.

Late watchers​

A fan-out stores nothing. A watcher receives what is published after its subscription has reached the publisher, and nothing from before. A watcher that starts late, or reconnects, misses the announcements in between. An announcement that matches no watcher is dropped without an error.

A service that announces to its own subscription loses nothing to this: its publisher reads that subscription's filter before the first send.

Slow watchers​

A watcher reads at most read_ahead deliveries ahead of its handler, 1000 by default. Past that it stops reading, and the socket buffers fill. The publisher then waits for the slowest watcher whose prefix matches. So a slow watcher slows the announcer down instead of losing announcements.

Retries on a fan-out​

A retry copy on a fan-out is a publish like any other: it reaches every subscription whose prefix matches. On a fan-out the service binds, a copy goes back to the whole audience the original had. A dead-letter name on the same endpoint matches the prefix too: jobs.dead arrives on jobs. and is handled again. A fan-out's dead letters go to an endpoint of their own.

Testing​

The announcer's test pushes a job and reads the announcement published on jobs.done. The watcher's test publishes jobs.done, jobs.failed and metrics.cpu as the announcer would, and checks that watch received the first two. The in-process mode applies the prefix rule; late joiners and back-pressure exist only on real sockets. Testing a ZeroMQ service covers both modes.