Commitable buffer

buffer.rb

class Buffer
  def initialize(size = 1_000, &blk)
    @size = size
    @callback = blk
    @queue = []
    @lock = Mutex.new
  end

  def <<(object)
    @lock.synchronize do
      @queue << object
      commit! if @queue.size >= @size
    end
    self
  end

  def commit!
    if @queue.any?
      @callback.call(@queue)
      @queue = []
    end
  end
end

usage.rb

# Set up buffer with commit callback as block
buffer = Buffer.new do |docs| 
  ElasticSearch.bulk(docs, :local)
  logger.increase_current_step
end

# Process products, keep adding results to buffer
products.each do |p|
  ...
  buffer << p
end

# Commit what is left in buffer
buffer.commit!