KafkaTopics

KafkaTopics reads a list of topics through one consumer and one group, and each delivery names its topic.

KafkaTopics::new takes the topics, and and_topic adds one more. The group settings are the ones KafkaTopic has: group, start, commit, assignment, lane_key and config. Every topic of the list shares the handler, so every topic carries the handler's payload type. Ctx<Topic> binds the topic a delivery came from. The subscription's name is the list joined with commas, orders,cancellations here. No one publish reaches a list again, so a registration over it names where a retry copy goes: .out_retry(policy).to(topic), or ToSourceTopic to send each copy back to the topic it came from. A registration that names neither stops the service at startup.

Example

Order events from two topics become one history, and each entry names the topic of its event.

Macros
Manual

examples/kafka_multi_topic.rs
/// Reads two topics through one consumer of the group `order-history`.
#[subscriber(
    KafkaTopics::new(["orders", "cancellations"]).group("order-history"),
    reply
)]
async fn record(event: &OrderEvent, Ctx(topic): Ctx<Topic>) -> HistoryEntry {
    HistoryEntry {
        id: event.id,
        topic: topic.to_string(),
    }
}

fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // No one topic reaches a topic set again, so the mount site says where a retried
        // event goes: back to the topic it came from.
        b.include(record)
            .out_retry(Publish::default())
            .transform(ToSourceTopic);
    })
}

examples/manual/kafka_multi_topic.rs
/// Answers each event with a history entry naming the topic it came from.
struct Record;

impl Handle<OrderEvent, HistoryEntry, (), KafkaContext> for Record {
    fn handle(
        &self,
        event: &OrderEvent,
        _outs: &(),
        ctx: &mut Context<'_, KafkaContext>,
    ) -> impl Future<Output = Result<HistoryEntry, HandlerOutcome>> {
        ready(Ok(HistoryEntry {
            id: event.id,
            topic: ctx.context(Topic).to_string(),
        }))
    }
}

fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // Two topics through one consumer of the group `order-history`.
        let events = KafkaTopics::new(["orders", "cancellations"]).group("order-history");
        // No one topic reaches a topic set again, so the mount site says where a retried
        // event goes: back to the topic it came from.
        b.include(
            subscriber("orders,cancellations", Record)
                .reply()
                .build()
                .map_source(|_| events),
        )
        .out_retry(Publish::default())
        .transform(ToSourceTopic);
    })
}

Example​

Example