Struct fluvio::Fluvio

pub struct Fluvio { /* fields omitted */ }

An interface for interacting with Fluvio streaming

Implementations

impl Fluvio

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

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?;

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

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?;

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

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_record("Hello, Fluvio!", 0).await?;

pub async fn partition_consumer<S: Into<String>>(
    &self,
    topic: S,
    partition: i32
) -> Result<PartitionConsumer, FluvioError>

Creates a new PartitionConsumer for the given topic and partition

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.

Example

let consumer_one = fluvio.partition_consumer("my-topic", 0).await?;
let consumer_two = fluvio.partition_consumer("my-topic", 1).await?;

let records_one = consumer_one.fetch(Offset::beginning()).await?;
let records_two = consumer_two.fetch(Offset::beginning()).await?;

pub async fn admin(&self) -> FluvioAdmin

Provides an interface for managing a Fluvio cluster

Example

let admin = fluvio.admin().await;

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.