Channels

Each channel reports its topic: the descriptor's for a subscription, the resolved destination for a publish.

The channel binding has one field, topic. A subscription by name, a KafkaTopic and a KafkaPartitions reader each report the one topic they read. A KafkaTopics set or pattern reads several topics, so its channel carries no binding.

A channel the service publishes to reports the topic the mount site resolved. That is a reply("dest") clause, a reply type's #[outgoing(name)], the name of a slot entry, or a declared dead-letter topic. A publish policy holds producer settings and no topic, so the policy itself adds nothing to the channel.

Example

The channels orders and ledger report the topics their descriptors read, and confirmations the topic the reply clause names. The channel of the audit and audit.eu set reports none.

Macros

examples/kafka_asyncapi.rs

#[subscriber(
    KafkaTopic::new("orders")
        .group("orders-svc")
        .config("client.id", "orders-reader"),
    reply("confirmations")
)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

#[subscriber(KafkaTopics::new(["audit", "audit.eu"]).group("audit-svc"))]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "audited");
    HandlerOutcome::ack()
}

#[subscriber(KafkaPartitions::new("ledger", [0, 1]))]
async fn ledger(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "booked");
    HandlerOutcome::ack()
}

Manual

examples/manual/kafka_asyncapi.rs

/// The topic `orders`, read in the group `orders-svc` by the client `orders-reader`.
fn orders() -> KafkaTopic {
    KafkaTopic::new("orders")
        .group("orders-svc")
        .config("client.id", "orders-reader")
}

/// Both audit topics, through one consumer in the group `audit-svc`.
fn audit_topics() -> KafkaTopics {
    KafkaTopics::new(["audit", "audit.eu"]).group("audit-svc")
}

/// The first two partitions of `ledger`, read apart from any group.
fn ledger_partitions() -> KafkaPartitions {
    KafkaPartitions::new("ledger", [0, 1])
}