Prelude

Each pattern's prelude offers its policy as Publish; ruststream_zeromq::prelude mounts several patterns under prefixed names.

A routes file on one pattern imports that pattern's prelude: queue::prelude, fanout::prelude or rpc::prelude. Each carries the core's prelude, so #[subscriber], RustStream, HandlerOutcome and Out come with it. Each adds ZmqEndpoint, the pattern's broker and its publish policy under the name every broker crate uses, Publish. rpc::prelude also brings RequestReply, the capability only ZmqRpc implements. So moving a service to another pattern changes the import line, and the mount site stays as it is.

A file that mounts more than one pattern imports ruststream_zeromq::prelude. One glob cannot carry three policies under one name, so this prelude names them ZmqQueuePublish, ZmqFanoutPublish and ZmqRpcPublish. It also carries the three brokers, RequestReply and the modules queue, fanout and rpc.

A handler body bounds its injected publisher with a core capability, such as Out<impl Publisher, Slot>, and imports ruststream::prelude::* alone. A ZeroMQ send takes the frames alone, so a handler body stays on the core's prelude, whichever pattern it runs on.

Broadcast with PUB/SUB mounts a queue and a fan-out in one file with the crate's prelude.

Example

A worker takes jobs from a queue and announces each finished job on a fan-out, so the routes file imports the crate's prelude.

Macros
Manual

examples/zmq_broker_prelude.rs
use ruststream_zeromq::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 {
    // Every listener that dials the fan-out hears about each finished job.
    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);
            },
        )
        .register_broker(events)
}

examples/manual/zmq_broker_prelude.rs
use ruststream_zeromq::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 {
    // Every listener that dials the fan-out hears about each finished job.
    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(subscriber("jobs", Work).reply().build())
                    .out_reply(announce);
            },
        )
        .register_broker(events)
}

Example​

Example