Method: Rdkafka::Consumer#each

Defined in:
lib/rdkafka/consumer.rb

#each {|message| ... } ⇒ nil

Poll for new messages and yield for each received one. Iteration will end when the consumer is closed.

If enable.partition.eof is turned on in the config this will raise an error when an eof is reached, so you probably want to disable that when using this method of iteration.

Yield Parameters:

  • message (Message) —

    Received message

Returns:

  • (nil)

Raises:


597
598
599
600
601
602
603
604
605
606
607
608
609
610
 
# File 'lib/rdkafka/consumer.rb', line 597

def each
  loop do
    message = poll(250)
    if message
      yield(message)
    else
      if closed?
        break
      else
        next
      end
    end
  end
end