autumn_plugin_kafka/
lib.rs

1//! Kafka producer and consumers for [Autumn](https://autumn-web.app) apps.
2//!
3//! # Quick start
4//!
5//! ```rust,no_run
6//! use autumn_plugin_kafka::{Consumer, HandlerError, KafkaPlugin, KafkaProducer, Message, Record};
7//! use autumn_web::prelude::*;
8//!
9//! #[derive(serde::Serialize, serde::Deserialize)]
10//! struct Order {
11//!     id: u64,
12//! }
13//!
14//! #[post("/orders/{id}")]
15//! async fn create(producer: KafkaProducer, Path(id): Path<u64>) -> AutumnResult<&'static str> {
16//!     producer
17//!         .send(Record::json("orders", &Order { id })?.with_key(id.to_string()))
18//!         .await?;
19//!     Ok("queued")
20//! }
21//!
22//! async fn on_order(msg: Message, _state: AppState) -> Result<(), HandlerError> {
23//!     let order: Order = msg.json()?;
24//!     tracing::info!(id = order.id, "order received");
25//!     Ok(())
26//! }
27//!
28//! #[autumn_web::main]
29//! async fn main() {
30//!     autumn_web::app()
31//!         .plugin(
32//!             KafkaPlugin::new().consumer(
33//!                 Consumer::new("orders", ["orders"])
34//!                     .group_id("billing")
35//!                     .handler(on_order)
36//!                     .dead_letter_topic("orders.dlq"),
37//!             ),
38//!         )
39//!         .routes(routes![create])
40//!         .run()
41//!         .await;
42//! }
43//! ```
44//!
45//! # Parts
46//!
47//! | Part | Function |
48//! |---|---|
49//! | [`KafkaPlugin`] | Installs all parts. Reads `[kafka]` at startup. |
50//! | [`KafkaConfig`] | The `[kafka]` config section. |
51//! | [`KafkaProducer`] | Sends records. It is a handler argument. |
52//! | [`Consumer`] | A consumer with an async handler. |
53//! | [`MemoryBroker`] | An in-memory broker for tests. |
54//! | [`RdKafkaBackend`] | The default client, `librdkafka`. |
55//!
56//! The plugin also adds the `kafka` health indicator and `kafka_*` metrics
57//! to `/actuator/prometheus`.
58//!
59//! # Delivery
60//!
61//! Consumers are at-least-once. The plugin commits an offset only after the
62//! handler completes, or after the dead-letter send completes.
63//! A handler can get the same message again. Make sure that this is safe.
64//!
65//! # Example
66//!
67//! `examples/orders.rs` uses all parts. CI runs its self-check.
68
69mod backend;
70mod config;
71mod consumer;
72mod error;
73mod health;
74mod memory;
75mod message;
76mod metrics;
77mod plugin;
78mod producer;
79mod rdkafka_backend;
80
81pub use backend::{Backend, ConsumerBackend, ConsumerSpec, ProducerBackend};
82pub use config::{ClientRole, ConsumerSettings, HealthSettings, KafkaConfig, ProducerSettings};
83pub use consumer::{
84    Consumer, DLQ_HEADER_CONSUMER, DLQ_HEADER_ERROR, DLQ_HEADER_OFFSET, DLQ_HEADER_PARTITION,
85    DLQ_HEADER_TOPIC, HandlerError,
86};
87pub use error::KafkaError;
88/// The future type of the backend traits.
89pub use futures::future::BoxFuture;
90pub use memory::MemoryBroker;
91pub use message::{Delivery, Header, Message, Record};
92pub use plugin::{KafkaPlugin, KafkaRuntime, PLUGIN_NAME};
93pub use producer::KafkaProducer;
94pub use rdkafka_backend::RdKafkaBackend;
autumn_plugin_kafka/lib.rs

autumn_plugin_kafka/
lib.rs
