consume/
consume.rs

1//! Fetch until the process is stopped. Broker on `KAFKA_BOOTSTRAP`.
2
3use partitionline::Consumer;
4
5#[tokio::main]
6async fn main() -> partitionline::Result<()> {
7    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
8    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
9    let mut consumer = Consumer::connect(bootstrap).await?;
10    consumer.assign_topic(topic, 0).await?;
11    loop {
12        for rec in consumer.fetch().await? {
13            println!(
14                "{}-{}@{} bytes={}",
15                rec.topic,
16                rec.partition,
17                rec.offset,
18                rec.value.as_ref().map(|v| v.len()).unwrap_or(0)
19            );
20        }
21    }
22}
consume/consume.rs

consume/
consume.rs
