EndpointRole

EndpointRole makes the side a type: ZmqQueue<Connect> dials, the parameter defaults to Bind, and a mount site writes it where it names the type.

ZmqEndpoint::bind returns a ZmqEndpoint<Bind>, and ZmqEndpoint::connect returns a ZmqEndpoint<Connect>. ZmqQueue and ZmqFanout carry the same Bind or Connect as their Role parameter, so ZmqQueue::new(ZmqEndpoint::connect(..)) is a ZmqQueue<Connect>. The parameter defaults to Bind, so ZmqQueue alone names a queue this process binds. ZmqRpc has no Role parameter.

A mount site writes the side only where it names the type: a Router for a queue it dials is a Router<ZmqQueue<Connect>>, and a test addresses that broker as tb.broker::<ZmqQueue<Connect>>(). The side decides where a retry copy can go, and the compiler holds a routes file to it. On ZmqQueue<Connect>, a router chain whose out_retry names no destination does not compile.

EndpointRole is the bound a helper writes to build a broker on either side. The trait is sealed: Bind and Connect are the whole set.

Example

A worker's routes on the queue it dials send their retry copies to a queue the worker binds. One helper builds both queues with the same read-ahead, whichever side the endpoint takes.

Macros
Manual

examples/zmq_broker_role.rs
/// The worker's routes on a queue it dials: the copies go to the queue `copies` publishes to.
fn routes(copies: Bound<ZmqQueue, Publish>) -> impl RouterDef<ZmqQueue<Connect>> {
    Router::<ZmqQueue<Connect>>::new()
        .include(work)
        .max_attempts(nonzero!(3u32))
        .out_retry(copies)
        .to("jobs")
        .build()
}

/// One queue setting for both sides of an endpoint.
fn queue<Role: EndpointRole>(endpoint: ZmqEndpoint<Role>) -> ZmqQueue<Role> {
    ZmqQueue::new(endpoint).read_ahead(nonzero!(64_usize))
}

#[ruststream::app]
fn app() -> impl App {
    let retries = queue(ZmqEndpoint::bind("tcp://0.0.0.0:5560")).bindable();
    let copies = retries.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(queue(ZmqEndpoint::connect("tcp://ventilator:5555")), |b| {
            b.include_router(routes(copies));
        })
        .with_broker(retries, |b| {
            b.include(work).max_attempts(nonzero!(3u32));
        })
}

examples/manual/zmq_broker_role.rs
/// The worker's routes on a queue it dials: the copies go to the queue `copies` publishes to.
fn routes(copies: Bound<ZmqQueue, Publish>) -> impl RouterDef<ZmqQueue<Connect>> {
    Router::<ZmqQueue<Connect>>::new()
        .include(subscriber("jobs", Work).build())
        .max_attempts(nonzero!(3u32))
        .out_retry(copies)
        .to("jobs")
        .build()
}

/// One queue setting for both sides of an endpoint.
fn queue<Role: EndpointRole>(endpoint: ZmqEndpoint<Role>) -> ZmqQueue<Role> {
    ZmqQueue::new(endpoint).read_ahead(nonzero!(64_usize))
}

fn app() -> impl App {
    let retries = queue(ZmqEndpoint::bind("tcp://0.0.0.0:5560")).bindable();
    let copies = retries.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(queue(ZmqEndpoint::connect("tcp://ventilator:5555")), |b| {
            b.include_router(routes(copies));
        })
        .with_broker(retries, |b| {
            b.include(subscriber("jobs", Work).build())
                .max_attempts(nonzero!(3u32));
        })
}

Example​

Example