consumer.rb

class Consumer
  class Fetcher
    def initialize(input, output)
      @input, @output = input, output
      
      @running = true
    end
    
    def run
      while @running
        if @input.empty?
          if @output.size < MAX_PENDING_OUTSPUTS
            fetch
          else
            sleep 1
          end
        else
          cmd, opts = @input.deq
          send("handle_#{cmd}", opts)
        end
      end
    end
    
    private
    
    def fetch
      # ...
      response = @broker.fetch_messages(topics: @topics)
      
      response.topics.each do |topic|
        topic.partitions.each do |batch|
          if batch.error_code == 0
            @output.enq([:batch, batch])
            # initiate a prefetch here...
          else
            @output.enq([:error, batch])
          end
        end
      end
    end
    
    def handle_shutdown
      @running = false
    end
    
    def handle_pause
    end
  end
  
  def each_batch(&block)
    inputs = brokers.map { Queue.new }
    output = Queue.new
    fetchers = brokers.zip(inputs).map {|broker, input| Fetcher.new(broker, input, output) }
    fetcher_threads = fetchers.map {|fetcher| Thread.new { fetcher.run } }
    
    loop do
      msg, batch = output.deq
      
      case msg
      when :batch
        yield batch
      when :error
        # ?
      end
    end
  end
end