ruststream_rdkafka/
lib.rs

1//! Apache Kafka broker for the [RustStream](https://github.com/powersemmi/ruststream) messaging
2//! framework, backed by [`rdkafka`] / librdkafka.
3//!
4//! # Transport model
5//!
6//! A subscription is one consumer joining one consumer group on one topic; [`KafkaTopic`]
7//! describes it, and the bare-string `#[subscriber("orders")]` form consumes the topic named
8//! `orders` through [`KafkaBroker::default_group`]. On the publish side
9//! [`OutgoingMessage::name`](ruststream::OutgoingMessage) is the destination topic, and a
10//! [`PARTITION_KEY_HEADER`] header becomes the record's native key, so Kafka itself keeps
11//! per-key ordering.
12//!
13//! Settlement follows Kafka's committed-position model instead of per-message frames; the
14//! [`Commit`] mode picks how:
15//!
16//! - [`Commit::Auto`] (the default): librdkafka auto-commit; `ack` and both `nack` forms are
17//!   advisory no-ops (the position is stored when a message is handed to the application, so
18//!   `nack(true)` does not cause a redelivery).
19//! - [`Commit::Tracked`]: precise at-least-once. `ack` settles its delivery and the stored
20//!   position advances across everything settled below it, staying correct under concurrent
21//!   handler lanes and across offset gaps (transaction markers, compacted topics).
22//!   `nack(false)` settles the offset (drop); `nack(true)` leaves it unsettled, so Kafka
23//!   redelivers from the committed position on the next fetch of the partition.
24//!
25//! Configuration delegates to librdkafka: unset options mean librdkafka defaults, and the raw
26//! `config(key, value)` passthroughs on the broker, the producer, and the descriptor reach
27//! every property this crate does not surface as a typed option.
28//!
29//! # Lazy startup
30//!
31//! [`KafkaBroker::new`] is synchronous and I/O-free, so a service composes with the synchronous
32//! `#[ruststream::app]` builder; the real network work happens in the idempotent async
33//! `Broker::connect`, called once by the runtime at startup. Publishers handed out before that
34//! resolve the shared connection on first use.
35//!
36//! [`rdkafka`]: https://docs.rs/rdkafka
37
38#![forbid(unsafe_code)]
39
40mod broker;
41mod convert;
42mod distribution;
43mod eos;
44mod error;
45mod message;
46mod publisher;
47mod retry;
48mod subscriber;
49mod topic;
50mod tracker;
51
52#[cfg(feature = "avro")]
53pub mod avro;
54pub mod context;
55#[cfg(feature = "protobuf")]
56pub mod protobuf;
57#[cfg(feature = "schema-registry")]
58pub mod schema_registry;
59#[cfg(feature = "testing")]
60pub mod testing;
61
62pub use broker::KafkaBroker;
63pub use distribution::RoundRobin;
64pub use eos::{EOS_SOURCE_HEADER, EosPipeline, EosReplies, SourceOffset};
65pub use error::KafkaError;
66pub use message::{KafkaMessage, PARTITION_HEADER, PARTITION_KEY_HEADER};
67pub use publisher::{KafkaPublisher, TransactionalPartitions};
68#[cfg(feature = "schema-registry")]
69pub use schema_registry::{
70    RegisteredSchema, SchemaFrame, SchemaRegistry, SchemaType, SubjectStrategy,
71};
72
73pub use retry::{
74    DLQ_SOURCE_OFFSET_HEADER, DLQ_SOURCE_PARTITION_HEADER, DLQ_SOURCE_TOPIC_HEADER,
75    RETRY_COUNT_HEADER, Retry,
76};
77pub use subscriber::KafkaSubscriber;
78pub use topic::{Assignment, Commit, KafkaTopic, LaneKey, StartOffset};
ruststream_rdkafka/lib.rs

ruststream_rdkafka/
lib.rs
