class

Kafka::Consumer

Inherits Reference / Object

Constructors

new(config : Hash(String, String))

Close the consumer and destroy the Kafka handle.

Calls the rd_kafka_consumer_close and rd_kafka_destroy C functions.

each(timeout = 250, raise_on_error = true, &)

Loops indefinitely calling #poll at the given interval timeout.

At the beginning of each loop, Fiber.yield is called allow other Fibers to run.

open?

Returns whether the consumer is open.

poll(timeout_ms : Int32, raise_on_error : Bool = true) : Message | Nil

Poll the consumer for messages or events.

Calls the rd_kafka_consumer_poll C function.

running?

stop(wait = false)

Stops the each loop at the next poll interval.

subscribe(*topics)

Subscribe to topics using balanced consumer groups.

Supports regex - start topic with '^'. For example:

consumer.subcribe("^foo") # will match any topics that starts with foo.

Raises a Kafka::ConsumerException when the subscribe fails.

Calls the rd_kafka_subscribe C function.