ruststream-rdkafka 0.5.1

Apache Kafka broker implementation for the RustStream messaging framework, backed by rdkafka / librdkafka.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

//! One subscription over several topics, and a pattern subscription: one consumer, one group,
//! one handler per subscription - all matched topics share the handler's payload type, and
//! each delivery still reports the topic it came from.
//!
//! ```text
//! just brokers-up
//! cargo run --example kafka_multi_topic -- run
//! ```

use ruststream::runtime::{App, AppInfo, HandlerResult, RustStream};
use ruststream::subscriber;
use ruststream_rdkafka::{KafkaBroker, KafkaTopic, StartOffset};
use serde::Deserialize;

#[derive(Debug, Deserialize)]
struct OrderEvent {
    id: u64,
}

// --8<-- [start:multi]
// `and_topic` adds topics to the same subscription: one consumer joins the group for both.
#[subscriber(KafkaTopic::new("orders").and_topic("cancellations").group("orders-svc"))]
async fn on_order_event(event: &OrderEvent) -> HandlerResult {
    println!("order event {}", event.id);
    HandlerResult::Ack
}
// --8<-- [end:multi]

// --8<-- [start:pattern]
// A `^`-anchored librdkafka regex subscribes to every matching topic; topics created later are
// picked up on the next metadata refresh.
#[subscriber(KafkaTopic::pattern("^audit\\..*").group("audit-svc").start(StartOffset::Earliest))]
async fn on_audit(event: &OrderEvent) -> HandlerResult {
    println!("audit event {}", event.id);
    HandlerResult::Ack
}
// --8<-- [end:pattern]

#[ruststream::app]
fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(on_order_event);
        b.include(on_audit);
    })
}