Retry

The Retry slot is the publisher a copy of a delivery leaves through; .out_retry(policy) replaces it and .to(name) names its destination.

The runtime publishes a copy of a delivery in two cases: a delayed retry_after on a broker without native delayed redelivery, and a delivery sent to its dead_letter destination. Every registration has a publisher for those copies, paired from the broker's default publish policy. .out_retry(policy) replaces it; it is the Out slot Retry, so .codec(..) and .transform(..) follow it as on any slot. A copy carries the delivery's own bytes, so no codec encodes it, and its PublishTransforms read the delivery being retried. A copy sent back to the subscription carries the RETRY_COUNT_HEADER, incremented. On the in-memory broker the delay is native, so only the dead-letter copy leaves through this slot. A subscription whose source descriptor declares AddressedCopies reports where it is reached again, and copies go there. One that declares NamedCopies reads many destinations and addresses none: a wildcard, a filter, a pattern. On it the mount site names the destination with .to(name), or a transform names one per delivery. Without either, a router chain does not compile, and a mount in a broker scope refuses to start. A name given with .to(name) is a channel the registration sends to, and the generated document reports it. A broker that moves a spent delivery itself declares BrokerMoves, and .out_retry(..) does not compile there.

Example

Macros

examples/settle_retry_destination.rs

/// Stamps every copy with the subscription its delivery came from.
struct RetriedFrom;

impl<C, Options> PublishTransform<ForReply<C>, Options> for RetriedFrom {
    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());
    }
}

#[subscriber("payments")]
async fn charge(payment: &Payment) -> HandlerOutcome {
    tracing::warn!(payment.id, "gateway refused the charge");
    HandlerOutcome::retry()
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(charge)
            .max_attempts(nonzero!(2u32))
            .dead_letter("payments.dead")
            // Replaces the broker's default publisher for the copies of this registration.
            .out_retry(Publish)
            .transform(RetriedFrom);
    })
}

Manual

examples/manual/settle_retry_destination.rs

/// Stamps every copy with the subscription its delivery came from.
struct RetriedFrom;

impl<C, Options> PublishTransform<ForReply<C>, Options> for RetriedFrom {
    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());
    }
}

struct Charge;

impl Handle<Payment> for Charge {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::warn!(payment.id, "gateway refused the charge");
        ready(Err(HandlerOutcome::retry()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("payments", Charge).build())
            .max_attempts(nonzero!(2u32))
            .dead_letter("payments.dead")
            // Replaces the broker's default publisher for the copies of this registration.
            .out_retry(Publish)
            .transform(RetriedFrom);
    })
}

A subscription that addresses nothing

Macros

examples/settle_retry_destination.rs

/// A subscription that reads every payment subject and addresses none of them, as a wildcard
/// or a topic filter does. `Copies = NamedCopies` says so.
#[derive(Debug, Clone)]
struct PaymentFilter {
    pattern: &'static str,
}

impl PaymentFilter {
    const fn new(pattern: &'static str) -> Self {
        Self { pattern }
    }
}

impl SubscriptionSource<ConnectedMemoryBroker> for PaymentFilter {
    type Subscriber = MemorySubscriber<Discarding>;
    type Copies = NamedCopies;

    fn name(&self) -> &str {
        self.pattern
    }

    async fn subscribe(
        self,
        connected: &ConnectedMemoryBroker,
    ) -> Result<Self::Subscriber, MemoryError> {
        MemoryPattern::new(self.pattern).subscribe(connected).await
    }
}

#[subscriber(PaymentFilter::new("payments.*"))]
async fn reconcile(payment: &Payment) -> HandlerOutcome {
    tracing::info!(payment.id, "not cleared yet");
    HandlerOutcome::retry_after(Duration::from_secs(5))
}

Manual

examples/manual/settle_retry_destination.rs

/// A subscription that reads every payment subject and addresses none of them, as a wildcard
/// or a topic filter does. `Copies = NamedCopies` says so.
#[derive(Debug, Clone)]
struct PaymentFilter {
    pattern: &'static str,
}

impl PaymentFilter {
    const fn new(pattern: &'static str) -> Self {
        Self { pattern }
    }
}

impl SubscriptionSource<ConnectedMemoryBroker> for PaymentFilter {
    type Subscriber = MemorySubscriber<Discarding>;
    type Copies = NamedCopies;

    fn name(&self) -> &str {
        self.pattern
    }

    async fn subscribe(
        self,
        connected: &ConnectedMemoryBroker,
    ) -> Result<Self::Subscriber, MemoryError> {
        MemoryPattern::new(self.pattern).subscribe(connected).await
    }
}

// What `subscriber(..)` accepts as a source: the descriptor passes through as it is.
impl IntoSource for PaymentFilter {
    type Source = Self;

    fn into_source(self) -> Self {
        self
    }
}

struct Reconcile;

impl Handle<Payment> for Reconcile {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(payment.id, "not cleared yet");
        ready(Err(HandlerOutcome::retry_after(Duration::from_secs(5))))
    }
}

Macros

examples/settle_retry_destination.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The descriptor addresses nothing, so the mount site names where a copy goes.
        b.include(reconcile).out_retry(Publish).to("payments.retry");
    })
}

/// The same registration with no destination named.
fn unnamed() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(reconcile);
    })
}

Manual

examples/manual/settle_retry_destination.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The descriptor addresses nothing, so the mount site names where a copy goes.
        b.include(subscriber(PaymentFilter::new("payments.*"), Reconcile).build())
            .out_retry(Publish)
            .to("payments.retry");
    })
}

/// The same registration with no destination named.
fn unnamed() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber(PaymentFilter::new("payments.*"), Reconcile).build());
    })
}