Struct fluvio::consumer::PartitionConsumer

pub struct PartitionConsumer<P = SpuPool> { /* private fields */ }

An interface for consuming events from a particular partition

Implementations

impl<P> PartitionConsumer<P>where P: SpuDirectory,

pub fn new( topic: String, partition: PartitionId, pool: Arc<P>, metrics: Arc<ClientMetrics> ) -> Self

pub fn topic(&self) -> &str

Returns the name of the Topic that this consumer reads from

pub fn partition(&self) -> PartitionId

Returns the ID of the partition that this consumer reads from

pub fn metrics(&self) -> Arc<ClientMetrics>

Return a shared instance of ClientMetrics

pub async fn stream( &self, offset: Offset ) -> Result<impl Stream<Item = Result<Record, ErrorCode>>>

Continuously streams events from a particular offset in the consumer’s partition

Streaming is one of the two ways to consume events in Fluvio. It is a continuous request for new records arriving in a partition, beginning at a particular offset. You specify the starting point of the stream using an Offset and periodically receive events, either individually or in batches.

Note this uses ConsumerRecord instead of batches

If you want more fine-grained control over how records are streamed, check out the stream_with_config method.

Example

use futures::StreamExt;
let mut stream = consumer.stream(Offset::beginning()).await?;
while let Some(Ok(record)) = stream.next().await {
    let key_str = record.get_key().map(|key| key.as_utf8_lossy_string());
    let value_str = record.get_value().as_utf8_lossy_string();
    println!("Got event: key={:?}, value={value_str}", key_str);
}

pub async fn stream_with_config( &self, offset: Offset, config: ConsumerConfig ) -> Result<impl Stream<Item = Result<Record, ErrorCode>>>

Continuously streams events from a particular offset in the consumer’s partition

Most of the time, you shouldn’t need to use a custom ConsumerConfig. If you don’t know what these settings do, try checking out the simpler PartitionConsumer::stream method that uses the default streaming settings.

Streaming is one of the two ways to consume events in Fluvio. It is a continuous request for new records arriving in a partition, beginning at a particular offset. You specify the starting point of the stream using an Offset and a ConsumerConfig, and periodically receive events, either individually or in batches.

Example

use futures::StreamExt;
// Use a custom max_bytes value in the config
let fetch_config = ConsumerConfig::builder()
    .max_bytes(1000)
    .build()?;
let mut stream = consumer.stream_with_config(Offset::beginning(), fetch_config).await?;
while let Some(Ok(record)) = stream.next().await {
    let key_str = record.get_key().map(|key| key.as_utf8_lossy_string());
    let value_str = record.get_value().as_utf8_lossy_string();
    println!("Got record: key={:?}, value={}", key_str, value_str);
}

pub async fn stream_batches_with_config( &self, offset: Offset, config: ConsumerConfig ) -> Result<impl Stream<Item = Result<Batch, ErrorCode>>>

Continuously streams batches of messages, starting an offset in the consumer’s partition

use futures::StreamExt;
// Use a custom max_bytes value in the config
let fetch_config = ConsumerConfig::builder()
    .max_bytes(1000)
    .build()?;
let mut stream = consumer.stream_batches_with_config(Offset::beginning(), fetch_config).await?;
while let Some(Ok(batch)) = stream.next().await {
    for record in batch.records() {
        let key_str = record.key().map(|key| key.as_utf8_lossy_string());
        let value_str = record.value().as_utf8_lossy_string();
        println!("Got record: key={:?}, value={}", key_str, value_str);
    }
}