KafkaPublish

KafkaPublish, Publish in the prelude, is the plain publish policy; every publish waits for the cluster's delivery report.

A mount site names it wherever a registration publishes: .out_reply(Publish::default()) for a reply, .out(Slot, Publish::default()) for a slot, .out_retry(Publish::default()) for retry copies. The policy holds no connection; the runtime pairs it with the connected broker at startup into a KafkaPublisher. A registration that names no policy for its reply publishes through this one. A publish returns once the cluster has acknowledged the record, as the producer's durability settings define it, or with the error it reported. librdkafka queues records locally and sends them in batches; a full queue makes a publish wait for room, which slows the handler to the pace of the cluster. queue_timeout bounds that wait: past it the publish fails with a queue-full error. A reply goes to one topic and does not wait for an answer from the reader. A synchronous exchange is a reply topic of the service's own plus a correlation header the requester sets and the responder copies to its reply. Per record, a publish carries a key or a partition; RoundRobin spreads replies over the partitions.

Example

Macros

examples/kafka_publishing_publish.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]),
        |b| {
            // A full local queue fails the reply after five seconds instead of holding the
            // handler until librdkafka has room again.
            b.include(confirm)
                .out_reply(Publish::default().queue_timeout(Duration::from_secs(5)));
        },
    )
}

Manual

examples/manual/kafka_publishing_publish.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]),
        |b| {
            let confirm = subscriber("orders", Confirm).map_source(|_| orders());
            // A full local queue fails the reply after five seconds instead of holding the
            // handler until librdkafka has room again.
            b.include(confirm.reply().build())
                .out_reply(Publish::default().queue_timeout(Duration::from_secs(5)));
        },
    )
}