Subscriptions

A subscription is a name, and the name is frame 0 of every message; #[subscriber("jobs")] is the whole declaration.

Every message travels as three frames, and the first one holds the name the publisher sent it under. That name is the whole declaration of a subscription, and the mount site carries every other setting. The manual form passes the name to subscriber(name, handler).

Only ZmqFanout reads the name: it hands a subscription every message whose name starts with the subscription's, as Prefixes describes. A ZmqQueue or ZmqRpc subscription takes every message its socket receives, whatever frame 0 says. So a second kind of work takes an endpoint of its own.

How many subscriptions a broker takes follows from the side. An endpoint this service binds is one socket, read by the subscription that bound it, and a second subscription there is refused when it opens. Subscriptions that dial are sockets of their own, so several of them dial one peer.

Nothing is stored on the way. A subscription that attaches late has no backlog to read, a restart replays nothing, and .start_at(..) does not compile on these brokers.

Example

A worker binds tcp://0.0.0.0:5555 and names its subscription jobs. A peer that pushes a message under another name to that endpoint reaches the same handler.

Macros

examples/zmq_subscriptions_name.rs

use ruststream_zeromq::queue::prelude::*;

#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "working on the job");
    HandlerOutcome::ack()
}

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

Manual

examples/manual/zmq_subscriptions_name.rs

use ruststream_zeromq::queue::prelude::*;

struct Work;

impl Handle<Job> for Work {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(job.id, "working on the job");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
        |b| {
            b.include(subscriber("jobs", Work).build());
        },
    )
}