Prelude

One glob, ruststream_rdkafka::prelude, brings the core prelude, the descriptors, the policies under concept names and the context keys.

use ruststream_rdkafka::prelude::*; is the one import a Kafka service writes. It carries the core's own prelude, so #[subscriber], RustStream, HandlerOutcome and Out come with it. It adds KafkaBroker, the descriptors KafkaTopic, KafkaTopics and KafkaPartitions with their settings (Commit, StartOffset, Assignment, LaneKey), the reply transforms RoundRobin and ToSourceTopic, and the context keys Topic, Partition, Position, Source and SeekHandle.

The publish policies come under the names every broker crate uses: Publish is KafkaPublish, TransactionalPublish is KafkaTransactionalPublish, PartitionedPublish is KafkaPartitionedPublish, and EosPublish is KafkaEosPublish. So a mount site reads the same on every broker. Two broker preludes in one file collide on these names; the prefixed types at the crate root tell them apart.

A handler body bounds its slots with capability traits, such as Out<impl TransactionalPublisher>, and names no broker type. The one exception is a body that sets a per-record setting: partition(..) needs KafkaPublishSteps in scope and the slot bounded Out<impl Publisher<Options = KafkaOptions>, Marker>, both from this prelude.

Example

The descriptor, its settings and the reply policy all come from the one glob.

Macros

examples/kafka_broker_prelude.rs

use ruststream_rdkafka::prelude::*;

#[subscriber(
    KafkaTopic::new("orders")
        .group("orders-svc")
        .commit(Commit::Tracked)
        .start(StartOffset::Earliest),
    reply
)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

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

Manual

examples/manual/kafka_broker_prelude.rs

use ruststream_rdkafka::prelude::*;

struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        ready(Ok(Confirmation { id: order.id }))
    }
}

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(|_| {
                KafkaTopic::new("orders")
                    .group("orders-svc")
                    .commit(Commit::Tracked)
                    .start(StartOffset::Earliest)
            });
            b.include(confirm.reply().build())
                .out_reply(Publish::default());
        },
    )
}