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(())
}
source

pub async fn consumer_offsets(&self) -> Result<Vec<ConsumerOffset>>

Returns all consumers offsets that currently available in the cluster.

source

pub async fn delete_consumer_offset( &self, consumer_id: impl Into<String>, replica_id: impl Into<ReplicaKey>, ) -> Result<()>

Delete a consumer offset for the given name and the replica.

source

pub async fn admin(&self) -> FluvioAdmin

Provides an interface for managing a Fluvio cluster

§Example
let admin = fluvio.admin().await;
source

pub fn platform_version(&self) -> &Version

Reports the Platform Version of the connected cluster.

The “Platform Version” is the value of the VERSION file when the cluster components were compiled, and is a semver value.

source

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

Auto Trait Implementations§

§

impl !Freeze for Fluvio

§

impl !RefUnwindSafe for Fluvio

§

impl Send for Fluvio

§

impl Sync for Fluvio

§

impl Unpin for Fluvio

§

impl !UnwindSafe for Fluvio

Blanket Implementations§

source§

impl<T> Any for T
where T: 'static + ?Sized,

source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
source§

impl<T> Borrow<T> for T
where T: ?Sized,

source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
source§

impl<T> From<T> for T

source§

fn from(t: T) -> T

Returns the argument unchanged.

source§

impl<T> Instrument for T

source§

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
source§

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
source§

impl<T, U> Into<U> for T
where U: From<T>,

source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

§

type Error = Infallible

The type returned in the event of a conversion error.
source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
source§

impl<T> WithSubscriber for T

source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
source§

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more
source§

impl<T> AsyncConnector for T
where T: Send + Sync,