Struct Consumer

pub struct Consumer<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

Implementations

impl<T, F> Consumer<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, Error>

Creates a new Consumer from group id and bootstrap servers

Arguments

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

Example

use simple_kafka::{Consumer};
let consumer = Consumer::from("group_id", "localhost:9092").unwrap();

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

Creates a new Consumer from consumer options

Arguments

• options - A ConsumerOptions struct that holds the consumer options

Example

use simple_kafka::{Consumer, 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 = Consumer::with_options(options).unwrap();

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

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 mut consumer = Consumer::from("group_id", "localhost:9092");
let handler = consumer.subscribe_to_topic("topic".to_string(), |data: Data, medatad: Metadata| async move {
   info!("data: {:?}, metadata: {:?}", data, medatad);
});
handler.await;