Broker

Three brokers, one per socket pattern: new records the endpoint with no I/O, and the runtime connects each at startup.

ZeroMQ has no server in the middle. Two processes talk directly, over a socket pair the pattern names. The pattern decides who receives a message, and the endpoint decides which side listens.

Broker	Sockets	Who receives a message	Publish policy
`ZmqQueue`	PUSH/PULL	one of the competing consumers, in turn	`ZmqQueuePublish`
`ZmqFanout`	PUB/SUB	every subscription whose name is a prefix of the message's	`ZmqFanoutPublish`
`ZmqRpc`	DEALER/ROUTER	the responder, and its answer the peer that asked	`ZmqRpcPublish`

Each broker's new takes a ZmqEndpoint and opens nothing, so the whole service fits the synchronous #[ruststream::app] builder. At startup the runtime connects it and receives its connected form, such as ConnectedZmqQueue. The connect checks the endpoint's scheme and opens no socket: each subscription and each publisher attaches its own socket when it first opens, as Startup describes. At shutdown the runtime consumes the connected form, and a publisher handed out earlier reports an error from then on, as Shutdown describes.

Each pattern's publish policy is also the default of its connected form. So a reply handler mounted with no publish position answers through its own pattern. Each pattern's prelude offers its policy as Publish, so moving a service to another pattern changes its import line and leaves the mount site as it is.

Getting started runs a first worker on a queue and feeds it from a Python peer.

Example

A worker stage of a pipeline binds tcp://0.0.0.0:5555, and any peer that pushes the three frames feeds it. cargo run --example zmq_pipeline --features testing -- run starts it.

Macros
Manual

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

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

examples/zmq_pipeline.rs
#[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(handle);
        },
    )
}

examples/manual/zmq_pipeline.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(()))
    }
}

examples/manual/zmq_pipeline.rs
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());
        },
    )
}

Example​

Example