Patterns

KafkaTopics::pattern reads every topic a librdkafka regex starting with ^ matches, those created later included.

The leading ^ is how librdkafka tells a pattern from a topic name. A pattern without it stops the service at startup, because librdkafka would subscribe to it as a literal name. and_pattern adds another pattern, and and_topic adds a literal topic to the same subscription. The consumer finds a topic created after startup on its next metadata refresh, and the group then assigns its partitions. A new topic is read from where the start offset puts it. The subscription's name is the pattern itself. A delivery names the topic it came from through Ctx<Topic>, and a retry copy needs a destination named at the mount site, as on KafkaTopics.

Example

Every topic under audit. is archived; auditing is not.

Macros

examples/kafka_multi_topic.rs

/// Reads every topic whose name starts with `audit.`, those created later included.
#[subscriber(KafkaTopics::pattern("^audit\\..*").group("audit-archive"))]
async fn archive(event: &OrderEvent, Ctx(topic): Ctx<Topic>) -> HandlerOutcome {
    tracing::info!(event.id, %topic, "archived");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        // A retried event goes back to the topic it came from.
        b.include(archive)
            .out_retry(Publish::default())
            .transform(ToSourceTopic);
    })
}

Manual

examples/manual/kafka_multi_topic.rs

/// Archives an event, logging the topic it came from.
struct Archive;

impl Handle<OrderEvent, (), (), KafkaContext> for Archive {
    fn handle(
        &self,
        event: &OrderEvent,
        _outs: &(),
        ctx: &mut Context<'_, KafkaContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(event.id, topic = %ctx.context(Topic), "archived");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        // Every topic whose name starts with `audit.`, those created later included.
        let audit = KafkaTopics::pattern("^audit\\..*").group("audit-archive");
        // A retried event goes back to the topic it came from.
        b.include(
            subscriber("^audit\\..*", Archive)
                .build()
                .map_source(|_| audit),
        )
        .out_retry(Publish::default())
        .transform(ToSourceTopic);
    })
}