roundtrip/
roundtrip.rs

1//! Produce then fetch one record.
2
3use std::time::Duration;
4
5use partitionline::{Consumer, ConsumerConfig, ProduceRecord, Producer, ProducerConfig};
6
7#[tokio::main]
8async fn main() -> partitionline::Result<()> {
9    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
10    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
11    let payload = b"live-roundtrip";
12
13    let mut pcfg = ProducerConfig::bootstrap([bootstrap.clone()]);
14    pcfg.linger = Duration::ZERO;
15    let producer = Producer::new(pcfg).await?;
16    let md = producer
17        .send(ProduceRecord::to(topic.clone()).value(&payload[..]))
18        .await?;
19    producer.close().await?;
20
21    let mut ccfg = ConsumerConfig::bootstrap([bootstrap]);
22    ccfg.max_wait_ms = 1000;
23    let mut consumer = Consumer::new(ccfg).await?;
24    consumer.assign(topic, md.partition, md.offset).await?;
25    let recs = consumer.fetch().await?;
26    let rec = recs
27        .iter()
28        .find(|r| r.offset == md.offset)
29        .ok_or_else(|| partitionline::Error::protocol("record not fetched"))?;
30    if rec.value.as_deref() != Some(&payload[..]) {
31        return Err(partitionline::Error::protocol("payload mismatch"));
32    }
33    println!(
34        "ok {}-{}@{} bytes={}",
35        rec.topic,
36        rec.partition,
37        rec.offset,
38        rec.value.as_ref().map(|v| v.len()).unwrap_or(0)
39    );
40    Ok(())
41}
roundtrip/roundtrip.rs

roundtrip/
roundtrip.rs
