Skip to main content

out_retry

out_retry(Publish) replaces the registration's retry publisher, and the steps after it, such as a transform, apply to every deferred copy.

The copies of a registration leave through its Retry slot. Every registration has a publisher there already, so a retry_after works with no out_retry at the mount site. out_retry names the publisher explicitly, and on this broker it is Publish, the crate's publish policy. The steps after it are the slot's steps. A PublishTransform over ForReply<C> reads the delivery being retried and edits each copy, here to add a header. It applies to every copy the registration publishes: the deferred copy, the copy an immediate retry becomes under a cap, and the dead-letter copy. The transform declares Destination = Reads: the copy goes where the runtime sends it, and a transform that declares Names does not compile on this position. The copy travels as the bytes that arrived, so no codec applies to it.

Example​

The transform stamps every copy with the subscription it came from:

examples/amqp_retry.rs
/// Marks every copy with the subscription the delivery came from, so a redelivery is
/// recognisable downstream. A transform on the retry position reads the delivery being retried.
struct Retried;

impl<C, Options> PublishTransform<ForReply<C>, Options> for Retried {
    type Destination = Reads;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        out.headers_mut()
            .insert("x-retried-from", cx.name().to_owned());
    }
}

The mount site names the publisher and the transform after it:

examples/amqp_retry.rs
#[ruststream::app]
fn app() -> impl App {
    let broker = AmqpBroker::new("amqp://artemis:artemis@localhost:5672");
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        // Five deliveries of one payment, the first included; the fifth failure goes to
        // "payments.dead" instead. Every copy leaves through `Publish`, stamped by `Retried`.
        b.include(charge)
            .max_attempts(nonzero!(5u32))
            .dead_letter("payments.dead")
            .out_retry(Publish)
            .transform(Retried);
        b.include(on_dead_letter);
    })
}