connect

ZmqEndpoint::connect makes this process dial: dialing subscriptions share a broker, and a broker that only publishes sends as usual.

Each subscription that dials is a socket of its own on the peer's endpoint. So one broker takes several of them: two ZmqQueue workers dialing one ventilator split its stream, and two ZmqFanout watchers dialing one publisher each receive their own prefix.

A subscription that dials reads from the sending end of its pattern, a ventilator's PUSH socket or a publisher's PUB socket, and that peer takes nothing. So once a subscription of this service has dialed the endpoint, a publish through the same ZmqQueue or ZmqFanout is refused with ZmqError::Send, naming the subscription. For the same reason a dialing subscription's retry copies go where its mount site names.

A broker that only publishes, such as a worker's link to the sink that collects its results, dials and sends as usual.

Pipelines with PUSH/PULL runs workers that dial both the ventilator and the sink.

Example

A pipeline worker dials both ends: the ventilator it pulls jobs from and the sink it pushes results to. Every copy of the worker runs the same app, so the service scales by starting more of them. The results and the retry copies of failed jobs leave for the sink.

Macros
Manual

examples/zmq_broker_connect.rs
use ruststream_zeromq::queue::prelude::*;

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

#[ruststream::app]
fn app() -> impl App {
    // The sink listens; this worker dials it and only publishes there.
    let sink = ZmqQueue::new(ZmqEndpoint::connect("tcp://sink:5556")).bindable();
    let results = sink.bind(Publish);
    let failed = sink.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker_labeled(
            "ventilator",
            ZmqQueue::new(ZmqEndpoint::connect("tcp://ventilator:5555")),
            |b| {
                // The ventilator only pushes: results and retry copies leave for the sink.
                b.include(work)
                    .out_reply(results)
                    .out_retry(failed)
                    .to("jobs.failed");
            },
        )
        .with_broker_labeled("sink", sink, |_b| {})
}

examples/manual/zmq_broker_connect.rs
use ruststream_zeromq::queue::prelude::*;

struct Work;

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

fn app() -> impl App {
    // The sink listens; this worker dials it and only publishes there.
    let sink = ZmqQueue::new(ZmqEndpoint::connect("tcp://sink:5556")).bindable();
    let results = sink.bind(Publish);
    let failed = sink.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker_labeled(
            "ventilator",
            ZmqQueue::new(ZmqEndpoint::connect("tcp://ventilator:5555")),
            |b| {
                // The ventilator only pushes: results and retry copies leave for the sink.
                b.include(subscriber("jobs", Work).reply().build())
                    .out_reply(results)
                    .out_retry(failed)
                    .to("jobs.failed");
            },
        )
        .with_broker_labeled("sink", sink, |_b| {})
}

Example​

Example