lib/poseidon/consumer_group.rb
# Initializes an infinite fetch loop. This method blocks!
#
# Will wait for `loop_delay` seconds after each failed fetch. This may happen when there is
# no new data or when the consumer hasn't claimed any partitions.
#
# SPECIAL ATTENTION:
# When 'breaking out' of the loop, you must do it before processing the messages, as the
# the last offset will not be committed. Please see examples below.
#
# @yield [partition, messages] The processing block
# @yieldparam [Integer] partition The source partition, may be -1 if no partitions are claimed
# @yieldparam [Array<Message>] messages The fetched messages
# @yieldreturn [Boolean] return false to prevent auto-commit
#
# @param [Hash] opts
# @option opts [Boolean] :commit Automatically commit consumed offset (default: true)
# @option opts [Boolean] :loop_delay Delay override in seconds after unsuccessful fetch.
#
# @example
#
#   group.fetch_loop do |n, messages|
#     puts "Fetched #{messages.size} messages for partition #{n}"
#   end
#   puts "Done" # => this code is never reached
#
# @example Stopping the loop (wrong)
#
#   counts = Hash.new(0)
#   group.fetch_loop do |n, messages|
#     counts[n] += messages.size
#     puts "Status: #{counts.inspect}"
#     break if counts[0] > 100
#   end
#   puts "Result: #{counts.inspect}"
#   puts "Offset: #{group.offset(0)}"
#
#   # Output:
#   # Status: {0=>30}
#   # Status: {0=>60}
#   # Status: {0=>90}
#   # Status: {0=>120}
#   # Result: {0=>120}
#   # Offset: 90      # => Last offset was not committed!
#
# @example Stopping the loop (correct)
#
#   counts = Hash.new(0)
#   group.fetch_loop do |n, messages|
#     break if counts[0] > 100
#     counts[n] += messages.size
#     puts "Status: #{counts.inspect}"
#   end
#   puts "Result: #{counts.inspect}"
#   puts "Offset: #{group.offset(0)}"
#
#   # Output:
#   # Status: {0=>30}
#   # Status: {0=>60}
#   # Status: {0=>90}
#   # Status: {0=>120}
#   # Result: {0=>120}
#   # Offset: 120
#
# @api public
Initializes an infinite fetch loop. This method blocks.
Will wait for `loop_delay` seconds after each failed fetch. This may happen when there is
no new data or when the consumer hasn't claimed any partitions.

SPECIAL ATTENTION:
When 'breaking out' of the loop, you must do it before processing the messages, as the
the last offset will not be committed. Please see examples below.
Initializes an infinite fetch loop . This method blocks . Will wait for ` loop_delay ` seconds after each failed fetch . This may happen when there is no new data or when the consumer hasn ' t claimed any partitions . SPECIAL ATTENTION : When ' breaking out ' of the loop you must do it before processing the messages as the the last offset will not be committed . Please see examples below .
def fetch_loop(opts = {})
    delay = opts[:loop_delay] || options[:loop_delay] || DEFAULT_LOOP_DELAY
    loop do
      mp = false
      ok = fetch(opts) do |n, messages|
        mp = !messages.empty?
        yield n, messages
      end
      unless ok
        yield -1, []
      end
      unless ok && mp
        sleep delay
      end
    end
  end
def fetch_loop ( opts = { } ) delay = opts [ :loop_delay ] || options [ :loop_delay ] || DEFAULT_LOOP_DELAY loop do mp = false ok = fetch ( opts ) do | n , messages | mp = ! messages . empty? yield n , messages end unless ok yield - 1 , [ ] end unless ok && mp sleep delay end end end
Initializes an infinite fetch loop.
Initializes an infinite fetch loop .
# Yield over an empty array if nothing claimed, # to allow user to e.g. break out of the loop # Sleep if either not claimes or nothing returned
return false to prevent auto-commit
return false to prevent auto - commit
Boolean
opts
:loop_delay Delay override in seconds after unsuccessful fetch.
: loop_delay Delay override in seconds after unsuccessful fetch .
]
<not_specific>
The processing block
The processing block
<not_specific>
None
example
group.fetch_loop do |n, messages| puts "Fetched #{messages.size} messages for partition #{n}" end puts "Done" # => this code is never reached
group . fetch_loop do |n messages| puts " Fetched # { messages . size } messages for partition # { n } " end puts " Done " # = > this code is never reached
example
Stopping the loop (wrong) counts = Hash.new(0) group.fetch_loop do |n, messages| counts[n] += messages.size puts "Status: #{counts.inspect}" break if counts[0] > 100 end puts "Result: #{counts.inspect}" puts "Offset: #{group.offset(0)}" # Output: # Status: {0=>30} # Status: {0=>60} # Status: {0=>90} # Status: {0=>120} # Result: {0=>120} # Offset: 90 # => Last offset was not committed!
Stopping the loop ( wrong ) counts = Hash . new ( 0 ) group . fetch_loop do |n messages| counts [ n ] + = messages . size puts " Status : # { counts . inspect } " break if counts [ 0 ] > 100 end puts " Result : # { counts . inspect } " puts " Offset : # { group . offset ( 0 ) } " # Output : # Status : { 0 = > 30 } # Status : { 0 = > 60 } # Status : { 0 = > 90 } # Status : { 0 = > 120 } # Result : { 0 = > 120 } # Offset : 90 # = > Last offset was not committed!
example
Stopping the loop (correct) counts = Hash.new(0) group.fetch_loop do |n, messages| break if counts[0] > 100 counts[n] += messages.size puts "Status: #{counts.inspect}" end puts "Result: #{counts.inspect}" puts "Offset: #{group.offset(0)}" # Output: # Status: {0=>30} # Status: {0=>60} # Status: {0=>90} # Status: {0=>120} # Result: {0=>120} # Offset: 120
Stopping the loop ( correct ) counts = Hash . new ( 0 ) group . fetch_loop do |n messages| break if counts [ 0 ] > 100 counts [ n ] + = messages . size puts " Status : # { counts . inspect } " end puts " Result : # { counts . inspect } " puts " Offset : # { group . offset ( 0 ) } " # Output : # Status : { 0 = > 30 } # Status : { 0 = > 60 } # Status : { 0 = > 90 } # Status : { 0 = > 120 } # Result : { 0 = > 120 } # Offset : 120
api
None