async fn consume(brokers: &str, group_id: &str, topics: &[&str]) {
    let consumer: StreamConsumer<DefaultConsumerContext> = ClientConfig::new()
        .set("group.id", group_id)
        .set("bootstrap.servers", brokers)
        .set("enable.partition.eof", "false")
        .set("session.timeout.ms", "6000")
        .set("enable.auto.commit", "true")
        //.set("statistics.interval.ms", "30000")
        //.set("auto.offset.reset", "smallest")
        .set_log_level(RDKafkaLogLevel::Debug)
        .create_with_context(DefaultConsumerContext)
        .expect("Consumer creation failed");

    consumer
        .subscribe(&topics.to_vec())
        .expect("Can't subscribe to specified topics");

    // consumer.stream() returns a stream. The stream can be used ot chain together expensive steps,
    // such as complex computations on a thread pool or asynchronous IO.
    let mut message_stream = consumer.stream();

    while let Some(message) = message_stream.next().await {
        match message {
            Err(e) => println!("Kafka error: {}", e),
            Ok(m) => {
                let event = m.to_event().unwrap();
                println!("Received Event: {:#?}", event);

                consumer.commit_message(&m, CommitMode::Async).unwrap();
            }
        };
    }
}
// You need a running Kafka cluster to try out this example. 
// With docker: docker run --rm --net=host -e ADV_HOST=localhost -e SAMPLEDATA=0 lensesio/fast-data-dev
You need a running Kafka cluster to try out this example.
You need a running Kafka cluster to try out this example .
async fn consume(brokers: &str, group_id: &str, topics: &[&str]) {
    let consumer: StreamConsumer<DefaultConsumerContext> = ClientConfig::new()
        .set("group.id", group_id)
        .set("bootstrap.servers", brokers)
        .set("enable.partition.eof", "false")
        .set("session.timeout.ms", "6000")
        .set("enable.auto.commit", "true")
        //.set("statistics.interval.ms", "30000")
        //.set("auto.offset.reset", "smallest")
        .set_log_level(RDKafkaLogLevel::Debug)
        .create_with_context(DefaultConsumerContext)
        .expect("Consumer creation failed");
    consumer
        .subscribe(&topics.to_vec())
        .expect("Can't subscribe to specified topics");
    // consumer.stream() returns a stream. The stream can be used ot chain together expensive steps,
    // such as complex computations on a thread pool or asynchronous IO.
    let mut message_stream = consumer.stream();
    while let Some(message) = message_stream.next().await {
        match message {
            Err(e) => println!("Kafka error: {}", e),
            Ok(m) => {
                let event = m.to_event().unwrap();
                println!("Received Event: {:#?}", event);
                consumer.commit_message(&m, CommitMode::Async).unwrap();
            }
        };
    }
}
async fn consume ( brokers : & str , group_id : & str , topics : & [ & str ] ) { let consumer : StreamConsumer < DefaultConsumerContext > = ClientConfig :: new ( ) . set ( " " , group_id ) . set ( " " , brokers ) . set ( " " , " " ) . set ( " " , " " ) . set ( " " , " " ) //.set("statistics.interval.ms", "30000") //.set("auto.offset.reset", "smallest") . set_log_level ( RDKafkaLogLevel :: Debug ) . create_with_context ( DefaultConsumerContext ) . expect ( " " ) ; consumer . subscribe ( & topics . to_vec ( ) ) . expect ( " " ) ; // consumer.stream() returns a stream. The stream can be used ot chain together expensive steps, // such as complex computations on a thread pool or asynchronous IO. let mut message_stream = consumer . stream ( ) ; while let Some ( message ) = message_stream . next ( ) . await { match message { Err ( e ) => println ! ( " " e ) , Ok ( m ) => { let event = m . to_event ( ) . unwrap ( ) ; println ! ( " " event ) ; consumer . commit_message ( & m , CommitMode :: Async ) . unwrap ( ) ; } } ; } }
You need a running Kafka cluster to try out this example.
You need a running Kafka cluster to try out this example .
brokers
&str
group_id
&str
topics
&[&str]
brokers
&str
group_id
&str
topics
&[&str]