Struct fluvio::Fluvio

source ·

pub struct Fluvio { /* private fields */ }

Expand description

An interface for interacting with Fluvio streaming

Implementations§

source §

impl Fluvio

source

pub async fn connect() -> Result<Self>

Creates a new Fluvio client using the current profile from ~/.fluvio/config

If there is no current profile or the ~/.fluvio/config file does not exist, then this will create a new profile with default settings and set it as current, then try to connect to the cluster using those settings.

§Example

let fluvio = Fluvio::connect().await?;

source

pub async fn connect_with_config(config: &FluvioConfig) -> Result<Self>

Creates a new Fluvio client with the given configuration

§Example

use fluvio::config::ConfigFile;
let config_file = ConfigFile::load_default_or_new()?;
let config = config_file.config().current_cluster().unwrap();
let fluvio = Fluvio::connect_with_config(&config).await?;

source

pub async fn connect_with_connector( connector: DomainConnector, config: &FluvioConfig ) -> Result<Self>

source

pub async fn topic_producer( &self, topic: impl Into<String> ) -> Result<TopicProducer>

Creates a new TopicProducer for the given topic name

Currently, producers are scoped to a specific Fluvio topic. That means when you send events via a producer, you must specify which partition each event should go to.

§Example

let producer = fluvio.topic_producer("my-topic").await?;
producer.send(RecordKey::NULL, "Hello, Fluvio!").await?;

source

pub async fn topic_producer_with_config( &self, topic: impl Into<String>, config: TopicProducerConfig ) -> Result<TopicProducer>

Creates a new TopicProducer for the given topic name

Currently, producers are scoped to a specific Fluvio topic. That means when you send events via a producer, you must specify which partition each event should go to.

§Example

let config = TopicProducerConfigBuilder::default().batch_size(500).build()?;
let producer = fluvio.topic_producer_with_config("my-topic", config).await?;
producer.send(RecordKey::NULL, "Hello, Fluvio!").await?;

source

pub async fn partition_consumer( &self, topic: impl Into<String>, partition: PartitionId ) -> Result<PartitionConsumer>

👎Deprecated since 0.21.8: use consumer_with_config() instead

Creates a new PartitionConsumer for the given topic and partition

If you have a topic with multiple partitions, then in order to receive all of the events in all of the partitions, use consumer instead.

source

pub async fn consumer( &self, strategy: PartitionSelectionStrategy ) -> Result<MultiplePartitionConsumer>

👎Deprecated since 0.21.8: use consumer_with_config() instead

Creates a new MultiplePartitionConsumer

Currently, consumers are scoped to both a specific Fluvio topic and to a particular partition within that topic. That means that if you have a topic with multiple partitions, then in order to receive all of the events in all of the partitions, you will need to create one consumer per partition.

Records across different partitions are not guaranteed to be ordered.

§Example

source

pub async fn consumer_with_config( &self, config: ConsumerConfigExt ) -> Result<impl ConsumerStream<Item = Result<Record, ErrorCode>>>

Creates a new ConsumerStream instance.

The stream can read data from one topic partition or all partitions. Records across different partitions are not guaranteed to be ordered.

The ConsumerStream provides the offset management capabilities. If configured, it allows to store consumed offsets in the Fluvio cluster, and use it later to continue reading from the last seen record.

If the offset_consumer property of ConsumerConfigExt is specified, the Fluvio fetches the offset by id and starts the stream from the next available record. If the offset does not exist, the Fluvio creates it. To read all existing consumers offsets one could use Self::consumer_offsets() function, to delete - Self::delete_consumer_offset() function.

The Fluvio saves offsets once one called ConsumerStream::offset_commit() method followed by ConsumerStream::offset_flush(). There is support for auto-commits if crate::consumer::OffsetManagementStrategy::Auto is used.

§Example

§Manually commit offsets

use fluvio::{
   consumer::{ConsumerConfigExtBuilder, ConsumerStream, OffsetManagementStrategy},
   Fluvio, Offset,
};
use futures_util::StreamExt;
async fn do_consume_with_manual_commits(fluvio: &Fluvio) -> anyhow::Result<()> {
   let mut stream = fluvio
       .consumer_with_config(
           ConsumerConfigExtBuilder::default()
               .topic("my-topic".to_string())
               .offset_consumer("my-consumer".to_string())
               .offset_start(Offset::beginning())
               .offset_strategy(OffsetManagementStrategy::Manual)
               .build()?,
       )
       .await?;
   while let Some(Ok(record)) = stream.next().await {
       println!("{}", String::from_utf8_lossy(record.as_ref()));
       stream.offset_commit()?;
       stream.offset_flush().await?;
   }
   Ok(())
}

§Auto-commits

use fluvio::{
   consumer::{ConsumerConfigExtBuilder, ConsumerStream, OffsetManagementStrategy},
   Fluvio, Offset,
};
use futures_util::StreamExt;
async fn do_consume_with_auto_commits(fluvio: &Fluvio) -> anyhow::Result<()> {
   let mut stream = fluvio
       .consumer_with_config(
           ConsumerConfigExtBuilder::default()
               .topic("my-topic".to_string())
               .offset_consumer("my-consumer".to_string())
               .offset_start(Offset::beginning())
               .offset_strategy(OffsetManagementStrategy::Auto)
               .build()?,
       )
       .await?;
   while let Some(Ok(record)) = stream.next().await {
       println!("{}", String::from_utf8_lossy(record.as_ref()));
   }
   Ok(())
}