partitionline 0.1.0

Pure-Rust Apache Kafka client and protocol implementation
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

//! Join a classic consumer group and print records.
//!
//! `KAFKA_TOPIC` may be a single name or a comma-separated list.

use partitionline::{ConsumerConfig, ConsumerGroup};

#[tokio::main]
async fn main() -> partitionline::Result<()> {
    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
    let group_id = std::env::var("KAFKA_GROUP").unwrap_or_else(|_| "partitionline".into());
    let topics: Vec<String> = std::env::var("KAFKA_TOPIC")
        .unwrap_or_else(|_| "partitionline".into())
        .split(',')
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(str::to_string)
        .collect();
    let mut group = ConsumerGroup::join_topics(
        ConsumerConfig::bootstrap([bootstrap]).max_wait_ms(500),
        group_id,
        topics,
    )
    .await?;
    loop {
        let recs = group.poll().await?;
        if recs.is_empty() {
            continue;
        }
        for rec in &recs {
            println!("{}-{}@{}", rec.topic, rec.partition, rec.offset);
        }
        group.commit_with_metadata(recs.next_offsets()).await?;
    }
}