KafkaBroker

KafkaBroker connects a service to a Kafka cluster; new only records the bootstrap servers, and the runtime connects it at startup.

KafkaBroker::new takes the bootstrap servers, each a host or host:port the client starts from. It does no I/O, so the whole service fits the synchronous #[ruststream::app] builder. The settings that follow it are builder steps too: a default consumer group, librdkafka properties through config, and the startup and shutdown bounds. A property the service leaves unset keeps the librdkafka default.

At startup the runtime connects the broker and receives a ConnectedKafkaBroker. The connect probes the cluster, so an unreachable cluster fails startup instead of the first publish. Every subscription and every publisher of the service opens against that connected form. At shutdown the runtime turns it into a ClosedKafkaBroker, after the producer has flushed what the handlers published.

#[subscriber("orders")] reads the topic orders in the broker's default group. Kafka reads a topic only through a consumer group, so a subscription with no group of its own on a broker with no default group fails startup. The descriptors that set a group, several topics or exact partitions per subscription are on the Subscriptions pages.

Example

The service reads the topic orders in the group orders-svc. cargo run --example kafka_quickstart --features testing -- run starts it against a cluster on localhost:9092.

Macros
Manual

examples/kafka_quickstart.rs
use ruststream_rdkafka::prelude::*;

#[subscriber("orders")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

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

examples/manual/kafka_quickstart.rs
use ruststream_rdkafka::prelude::*;

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 {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
        |b| {
            b.include(subscriber("orders", Place).build());
        },
    )
}

Example​

Example