Struct PausableConsumer

pub struct PausableConsumer<T, F>
where
    T: for<'b> Deserialize<'b>,
    F: Future<Output = ()> + Send + Sync + 'static,
{ /* private fields */ }

A Consumer that can be used to consume messages from kafka and has the ability to pause and resume

Implementations

impl<T, F> PausableConsumer<T, F>

where T: for<'a> Deserialize<'a>, F: Future<Output = ()> + Send + Sync + 'static,

pub fn from( group_id: &str, bootstrap_servers: &str, ) -> Result<(Self, Arc<Mutex<bool>>), Error>

Creates a new PausableConsumer from a group_id and bootstrap_servers

Arguments

• group_id - The group_id of the consumer
• bootstrap_servers - The comma separated bootstrap servers

Returns

A tuple of PausableConsumer and a Arc<Mutex>. The Arc<Mutex> is used to pause and resume the consumer

Example

use simple_kafka::{PausableConsumer};

let (consumer, paused) = PausableConsumer::from("group_id", "localhost:9092").unwrap();

pub fn with_options( options: ConsumerOptiopns<'_>, ) -> Result<(Self, Arc<Mutex<bool>>), Error>

Creates a new PausedConsumer from consumer options

Arguments

• options - A ConsumerOptions struct that holds the consumer options

Returns

A tuple of PausableConsumer and a Arc<Mutex>. The Arc<Mutex> is used to pause and resume the consumer

Example

use simple_kafka::{PausableConsumer, ConsumerOptiopns};

let options = ConsumerOptiopns {
    bootstrap_servers: "localhost:9092".to_string(),
    group_id: "group_id".to_string(),
    session_timeout_ms: "6000".to_string(),
    enable_auto_commit: true,
    enable_partition_eof: false,
};
let consumer = PausableConsumer::with_options(options)?;

pub async fn subscribe_to_topic<H>( &mut self, topic: &str, handler: H, ) -> Result<(), Error>

where H: Fn(KafkaResult<T>) -> F,

Subscribe to a given topic and calls the given function for each message

Arguments

• topic - The topic to subscribe to
• handler - The handler function that will be called for each message for the give topic

Example

use simple_kafka::{Consumer};
#[derive(Serialize, Deserialize, Debug)]
 struct Data {
    attra_one: String,
    attra_two: i8,
}

let consumer = Consumer::from("group_id", "localhost:9092");
let handler_1 = | data: Data, metadata: Metadata| async move {
    println!("Handler One ::: data: {:?}, metadata: {:?}", data, metadata);
};
consumer.subscribe_to_topic("topic_1".to_string(), handler_1).await.unwrap();

pub async fn pause(&self)

Pause the consumer. Consumer will not request new messages but will keep the connection to the broker alive This is useful when you want to pause the consumer for a while and resume it later without having to reconnect to the broker

Example

use simple_kafka::{Consumer};

#[tokio::main]
async fn main() {
    let (mut consumer, is_runnig)  = PausableConsumer::from("group_id", "localhost:9092").unwrap();
    let handler = consumer.subscribe_to_topic("topic".to_string(), |data: Data, medatad: Metadata| async move {
        info!("data: {:?}, metadata: {:?}", data, medatad);
    });
    consumer.pause().await;
}

pub async fn resume(&self)

Resume the consumer

Example

use simple_kafka::{Consumer};

#[tokio::main]
async fn main() {
    let (mut consumer, is_runnig)  = PausableConsumer::from("group_id", "localhost:9092").unwrap();
    let handler = consumer.subscribe_to_topic("topic".to_string(), |data: Data, medatad: Metadata| async move {
        info!("data: {:?}, metadata: {:?}", data, medatad);
    });
    consumer.pause().await;
    consumer.resume().await;
}