out_retry

out_retry(policy) replaces the publisher the retry copies leave through, so the copies take a transform or a codec of the mount site's choosing.

A Kinesis registration publishes its retry copies without it, through the broker's default KinesisPublish policy. out_retry names the policy for that position, and the chain after it takes the position's steps: .transform(..) and .codec(..). A transform on the retry position reads the delivery it answers, through a PublishContext, and runs on every copy. The copy carries the record's own bytes, so a codec named there encodes nothing. The copy keeps the record's partition key and lands on that key's shard.

Example

Every invoice copy says which stream it was deferred from.

Macros

examples/kinesis_settlement_out_retry.rs

/// Writes the stream a record was deferred from onto its retry copy.
struct DeferredFrom;

impl<Context, Options> PublishTransform<ForReply<Context>, Options> for DeferredFrom {
    type Destination = Reads;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, Context>,
    ) {
        out.headers_mut()
            .insert(DEFERRED_FROM, cx.name().to_owned());
    }
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        b.include(collect)
            .out_retry(Publish::default())
            .transform(DeferredFrom);
    })
}

Manual

examples/manual/kinesis_settlement_out_retry.rs

/// Writes the stream a record was deferred from onto its retry copy.
struct DeferredFrom;

impl<Context, Options> PublishTransform<ForReply<Context>, Options> for DeferredFrom {
    type Destination = Reads;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, Context>,
    ) {
        out.headers_mut()
            .insert(DEFERRED_FROM, cx.name().to_owned());
    }
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(KinesisStream::new("invoices"), Collect).build())
            .out_retry(Publish::default())
            .transform(DeferredFrom);
    })
}