Struct fluvio::consumer::MultiplePartitionConsumer

pub struct MultiplePartitionConsumer { /* private fields */ }

Implementations

impl MultiplePartitionConsumer

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

👎Deprecated since 0.21.8: use Fluvio::consumer_with_config() instead

Continuously streams events from a particular offset in the selected partitions

Streaming is one of the two ways to consume events in Fluvio. It is a continuous request for new records arriving in the selected partitions, 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.

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={}", key_str, value_str);
}

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

👎Deprecated since 0.21.8: use Fluvio::consumer_with_config() instead

Continuously streams events from a particular offset in the selected partitions

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 [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 the selected partitions, 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);
}

Trait Implementations

impl Clone for MultiplePartitionConsumer

fn clone(&self) -> MultiplePartitionConsumer

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.