fixed_queue_with_error.rb

require 'thread'
require 'logger'

POOL_SIZE = 10 # limit for HTTP parallel connections
QUEUE_SIZE = 5 # limit to consume only fixed amount of memory

queue = SizedQueue.new(QUEUE_SIZE)
logger = Logger.new($stdout)
END_MESSAGE = :done
ERROR_MESSAGE = :error

threads = POOL_SIZE.times.map do |t|
  Thread.new do
    10.times do |i|
      # imagine HTTP request here I want to do in parallel and pushing result to the queue
      if rand > 0.9
        raise "error"
      end
      queue.push(:hello)
      logger.info "T#{t} pushed message #{i}"
      sleep rand
    end
    queue.push(END_MESSAGE)
    logger.info "T#{t} done"
  rescue
    queue.push(ERROR_MESSAGE)
    logger.error "T#{t} errored"
  end
end

finished = 0
consumed = 0

begin
  loop do
    message = queue.pop
    if message == END_MESSAGE
      finished = finished + 1
      logger.info "done, finished #{finished} threads"
      break if finished == POOL_SIZE
    elsif message == ERROR_MESSAGE
      raise "queue problem"
    else
      # store message to DB
      sleep 0.01
      consumed = consumed + 1
      logger.info "pop, consumed #{consumed} messages"
    end
  end

  logger.info "all done, consumed #{consumed} messages, finished #{finished} threads"
rescue
  logger.error "error, consumed #{consumed} messages, finished #{finished} threads"
end