ToSourceTopic

ToSourceTopic sends each retry copy back to the topic its delivery arrived on, for subscriptions over several topics.

A KafkaTopic subscription reads one topic, and its retry copies go there. A KafkaTopics list or pattern and a KafkaPartitions reader have no single topic, so the mount site names one. ToSourceTopic is a transform of the Retry slot: .out_retry(Publish::default()).transform(ToSourceTopic). It reads each delivery's topic from its Kafka context, so a copy of an orders-us record goes back to orders-us. The context reaches the slot when the handler reads it: a Ctx<Topic> parameter, or a manual body typed over KafkaContext. A batch handler carries no per-record topic, so a batch registration names a fixed topic instead. .out_retry(Publish::default()).to(topic) sends every copy to that one topic; a pattern that matches it reads the copies back. A registration over several topics that names neither does not start.

Example

Macros

examples/kafka_settlement_source_topic.rs

/// One subscription reads both regions. Reading the topic is also what hands the Kafka context
/// to the retry position, where `ToSourceTopic` reads it.
#[subscriber(KafkaTopics::new(["orders-eu", "orders-us"]).group("orders-svc").commit(Commit::Tracked))]
async fn reserve(
    order: &Order,
    Ctx(topic): Ctx<Topic>,
    State(warehouse): State<Arc<Warehouse>>,
) -> HandlerOutcome {
    if let Err(err) = warehouse.reserve(order) {
        tracing::warn!(order.id, %topic, err, "warehouse unavailable, retrying");
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, %topic, "stock reserved");
    HandlerOutcome::ack()
}

Manual

examples/manual/kafka_settlement_source_topic.rs

/// One subscription reads both regions. A body typed over `KafkaContext` hands that context to
/// the retry position, where `ToSourceTopic` reads it.
struct Reserve;

impl Handle<Order, (), (), KafkaContext, AppState> for Reserve {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, KafkaContext, AppState>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let topic = ctx.context(Topic);
        if let Err(err) = ctx.state().warehouse.reserve(order) {
            tracing::warn!(order.id, %topic, err, "warehouse unavailable, retrying");
            return ready(Err(HandlerOutcome::retry()));
        }
        tracing::info!(order.id, %topic, "stock reserved");
        ready(Ok(()))
    }
}

fn regions() -> KafkaTopics {
    KafkaTopics::new(["orders-eu", "orders-us"])
        .group("orders-svc")
        .commit(Commit::Tracked)
}

Macros

examples/kafka_settlement_source_topic.rs

#[ruststream::app]
fn app() -> impl App<State = AppState> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(AppState::default()))
        .with_broker(broker, |b| {
            // A copy goes back to the topic its delivery arrived on.
            b.include(reserve)
                .max_attempts(nonzero!(3u32))
                .dead_letter("dlq.orders")
                .out_retry(Publish::default())
                .transform(ToSourceTopic);
            // Every copy goes to one topic. `.to(topic)` belongs to the publisher the copies
            // leave through, so it follows `out_retry`.
            b.include(post_entry)
                .max_attempts(nonzero!(3u32))
                .dead_letter("dlq.ledger")
                .out_retry(Publish::default())
                .to("ledger.retry");
        })
}

Manual

examples/manual/kafka_settlement_source_topic.rs

fn app() -> impl App<State = AppState> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(AppState::default()))
        .with_broker(broker, |b| {
            // A copy goes back to the topic its delivery arrived on.
            b.include(
                subscriber("orders-eu,orders-us", Reserve)
                    .map_source(|_| regions())
                    .build(),
            )
            .max_attempts(nonzero!(3u32))
            .dead_letter("dlq.orders")
            .out_retry(Publish::default())
            .transform(ToSourceTopic);
            // Every copy goes to one topic. `.to(topic)` belongs to the publisher the copies
            // leave through, so it follows `out_retry`.
            b.include(
                subscriber("^ledger\\..*", PostEntry)
                    .map_source(|_| ledgers())
                    .build(),
            )
            .max_attempts(nonzero!(3u32))
            .dead_letter("dlq.ledger")
            .out_retry(Publish::default())
            .to("ledger.retry");
        })
}

A fixed retry topic

Macros

examples/kafka_settlement_source_topic.rs

/// A pattern reads every ledger topic there is; its copies go to one retry topic the pattern
/// also matches, so they come back to this subscription.
#[subscriber(KafkaTopics::pattern("^ledger\\..*").group("ledger-svc").commit(Commit::Tracked))]
async fn post_entry(entry: &Entry, State(ledger): State<Arc<Ledger>>) -> HandlerOutcome {
    if let Err(err) = ledger.post(entry) {
        tracing::warn!(entry.order_id, err, "ledger unavailable, retrying");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/kafka_settlement_source_topic.rs

/// A pattern reads every ledger topic there is; its copies go to one retry topic the pattern
/// also matches, so they come back to this subscription.
struct PostEntry;

impl Handle<Entry, (), (), (), AppState> for PostEntry {
    fn handle(
        &self,
        entry: &Entry,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if let Err(err) = ctx.state().ledger.post(entry) {
            tracing::warn!(entry.order_id, err, "ledger unavailable, retrying");
            return ready(Err(HandlerOutcome::retry()));
        }
        ready(Ok(()))
    }
}

fn ledgers() -> KafkaTopics {
    KafkaTopics::pattern("^ledger\\..*")
        .group("ledger-svc")
        .commit(Commit::Tracked)
}