Prefixes

On ZmqFanout a subscription named orders receives orders.created too: the name is a prefix filter, the protocol's own rule.

A SUB socket subscribes with the subscription's name, and a message reaches it when frame 0 starts with that name. The comparison runs byte by byte and knows no separator. So orders takes orders.created and orders.shipped, and orders-archive as well. A name that ends with the separator, such as orders., takes the dotted family and leaves the lookalike out. A message that matches no subscription is dropped without an error.

The rule is the same on both sides. A subscription that binds the fan-out and one that dials a publisher filter alike, and each subscription that dials is a socket with its own filter. A message this service publishes to its own bound fan-out is matched the same way, so a name under the subscription's prefix comes back to it.

In the generated document, a fan-out publish channel reports nameFrame: the prefix a SUB peer subscribes with to receive it.

Broadcast with PUB/SUB runs a watcher that subscribes to every job announcement by prefix.

Example

The auditor reads every order event and leaves the archive records alone.

Macros

examples/zmq_subscriptions_prefixes.rs

use ruststream_zeromq::fanout::prelude::*;

/// Every order event, `orders.created`, `orders.shipped`, and no `orders-archive` record.
#[subscriber("orders.")]
async fn audit(event: &OrderEvent) -> HandlerOutcome {
    tracing::info!(event.order_id, "order event audited");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("auditor", "0.1.0")).with_broker(
        ZmqFanout::new(ZmqEndpoint::bind("tcp://0.0.0.0:5556")),
        |b| {
            b.include(audit);
        },
    )
}

Manual

examples/manual/zmq_subscriptions_prefixes.rs

use ruststream_zeromq::fanout::prelude::*;

/// Every order event, `orders.created`, `orders.shipped`, and no `orders-archive` record.
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 {
    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());
        },
    )
}