Subscriptions

A topic name in #[subscriber] reads that topic in the broker's default group; a descriptor carries every other setting.

#[subscriber("orders")] reads the topic orders. Kafka reads a topic only through a consumer group, and a bare name names none, so the group comes from KafkaBroker::default_group. A subscription with no group, on a broker with no default group, stops the service at startup with an error that names the subscription. Two subscriptions of one name and one group are two members of that group: each record goes to one of them. Every other setting of a subscription belongs to a descriptor: KafkaTopic for one topic, KafkaTopics for several topics or a pattern, and KafkaPartitions for named partitions. A descriptor goes where the name goes, as the attribute's first argument. The manual form passes the name to subscriber(name, body), and a descriptor joins the chain through map_source. A subscription is open once its group has assigned it its partitions, so everything published after startup reaches it. Startup waits for that assignment up to KafkaBroker::assignment_timeout, 60 seconds by default.

Example

The topic orders, read in the group orders-svc.

Macros

examples/kafka_subscriptions_name.rs

/// Reads the topic `orders`; the group comes from the broker.
#[subscriber("orders")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("orders-svc");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(place);
    })
}

Manual

examples/manual/kafka_subscriptions_name.rs

/// Reads the topic `orders`; the group comes from the broker.
struct Place;

impl Handle<Order> for Place {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order placed");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("orders-svc");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("orders", Place).build());
    })
}