out_retry

Where copies go follows the side: a bound subscription takes its own, and one that dials and every responder name a destination.

out_retry(policy) sets the publisher the retry copies leave through. A ZmqQueue or ZmqFanout subscription that binds its endpoint is an address of its own. Its copies go back to that endpoint, so the mount names no destination, and .out_retry(Publish) states the default.

A subscription that dials reads from a peer that only sends, so it has no address to send copies to. Its mount site names one with .out_retry(policy).to("name"). The policy is often a queue on another broker, which bindable() and .bind(Publish) turn into a publisher of this mount. A mount on a dialing subscription that names no destination is refused before the subscription opens, with an error naming the subscription. A Router chain that names none does not compile. Both hold for a mount that never asks for a retry, because the runtime pairs the publisher either way.

Every ZmqRpc responder names a destination too. A request has no address a copy could return to: answers travel to the identity of the peer that asked, and the reply publisher refuses a plain name.

Retries and dead letters mounts both forms, a worker that binds and a worker that dials.

Example

The worker dials a ventilator and sends its copies to a queue it binds itself, where the same handler takes them. The copies carry the count, so the cap of three holds across both subscriptions.

Macros
Manual

examples/zmq_settlement_out_retry.rs
#[ruststream::app]
fn app() -> impl App {
    let retries = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5560")).bindable();
    let copies = retries.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::connect("tcp://ventilator:5555")),
            |b| {
                // The ventilator takes nothing back, so the copies go to the queue bound below.
                b.include(handle)
                    .max_attempts(nonzero!(3u32))
                    .out_retry(copies)
                    .to("jobs");
            },
        )
        .with_broker(retries, |b| {
            // A bound queue takes its own copies back, so this mount names no destination.
            b.include(handle).max_attempts(nonzero!(3u32));
        })
}

examples/manual/zmq_settlement_out_retry.rs
fn app() -> impl App {
    let retries = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5560")).bindable();
    let copies = retries.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::connect("tcp://ventilator:5555")),
            |b| {
                // The ventilator takes nothing back, so the copies go to the queue bound below.
                b.include(subscriber("jobs", Work).build())
                    .max_attempts(nonzero!(3u32))
                    .out_retry(copies)
                    .to("jobs");
            },
        )
        .with_broker(retries, |b| {
            // A bound queue takes its own copies back, so this mount names no destination.
            b.include(subscriber("jobs", Work).build())
                .max_attempts(nonzero!(3u32));
        })
}

Example​

Example