Kafka bindings

With the asyncapi feature the generated document speaks Kafka: the server, the topic behind each channel, the group of each operation.

The core builds the AsyncAPI document from the app, before anything connects. This crate adds the specification's kafka binding, version 0.5.0, to the parts it knows about. A server reports the cluster's coordinates and its schema registry. A channel reports the topic behind it. A receive operation reports the consumer group it reads in.

Everything comes from the service's own declarations: the broker, the descriptors and the destinations the mount sites resolved. The document is published and shared, so no binding carries a credential. The binary #[ruststream::app] writes prints the document with asyncapi gen.

Example

The dependency line:

ruststream-rdkafka = { version = "0.7", features = ["asyncapi"] }

The service registers its broker under the label kafka, the key of its server in the document. cargo run --example kafka_asyncapi --features asyncapi,schema-registry -- asyncapi gen prints the document.

Macros

examples/kafka_asyncapi.rs

fn service(broker: KafkaBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "1.0.0")).with_broker_labeled("kafka", broker, |b| {
        b.include(confirm);
        b.include(audit);
        b.include(ledger);
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    let password = env::var("KAFKA_PASSWORD").unwrap_or_default();
    service(broker(&password))
}

Manual

examples/manual/kafka_asyncapi.rs

fn service(broker: KafkaBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "1.0.0")).with_broker_labeled("kafka", broker, |b| {
        let confirm = subscriber("orders", Confirm).map_source(|_| orders());
        b.include(confirm.reply().to("confirmations").build());
        b.include(
            subscriber("audit", Audit)
                .map_source(|_| audit_topics())
                .build(),
        );
        b.include(
            subscriber("ledger", Ledger)
                .map_source(|_| ledger_partitions())
                .build(),
        );
    })
}

fn app() -> impl App<State = ()> {
    let password = env::var("KAFKA_PASSWORD").unwrap_or_default();
    service(broker(&password))
}