ZmqFanout

ZmqFanout is PUB/SUB: every subscription whose name prefixes a message gets a copy, and one that attaches late misses what came before.

The subscription's name is its filter, and the filter is a prefix, the protocol's own rule. #[subscriber("orders")] receives orders.created and orders.shipped, and never payments.settled; the prefixes page covers the rule. A message that matches no subscription is dropped, and the publisher gets no error.

Nothing is stored. A subscription that attaches after a publisher has started misses what that publisher sent before it arrived. This is the slow joiner of ZeroMQ. A publisher that reaches a subscription of its own service reads that subscription's filter before its first send, so inside one service nothing is lost to it.

The policy is ZmqFanoutPublish, offered as Publish by fanout::prelude. A PUB socket is held back by its slowest matching subscriber, as read_ahead describes.

A retry copy on a fan-out the subscription binds goes back to that subscription's listener and reaches the audience the original had. A subscription that dials reads from a publisher that takes nothing back, so its mount site names where the copies go, as on ZmqQueue.

Broadcast with PUB/SUB announces finished jobs to watchers that subscribe by prefix.

Example

An auditor binds a fan-out, and the order services dial it. Its subscription on orders receives every order event, whatever comes after the prefix.

Macros
Manual

examples/zmq_broker_fanout.rs
use ruststream_zeromq::fanout::prelude::*;

#[subscriber("orders")]
async fn audit(event: &OrderEvent) -> HandlerOutcome {
    tracing::info!(event.order_id, "order event audited");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // The order services dial this endpoint and publish `orders.created`, `orders.shipped`, ...
    RustStream::new(AppInfo::new("auditor", "0.1.0")).with_broker(
        ZmqFanout::new(ZmqEndpoint::bind("tcp://0.0.0.0:5556")),
        |b| {
            b.include(audit);
        },
    )
}

examples/manual/zmq_broker_fanout.rs
use ruststream_zeromq::fanout::prelude::*;

struct Audit;

impl Handle<OrderEvent> for Audit {
    fn handle(
        &self,
        event: &OrderEvent,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(event.order_id, "order event audited");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    // The order services dial this endpoint and publish `orders.created`, `orders.shipped`, ...
    RustStream::new(AppInfo::new("auditor", "0.1.0")).with_broker(
        ZmqFanout::new(ZmqEndpoint::bind("tcp://0.0.0.0:5556")),
        |b| {
            b.include(subscriber("orders", Audit).build());
        },
    )
}

Example​

Example