array_blocking_queue.rb

class ArrayBlockingQueue
  attr_reader :capacity, :queue

  def initialize(capacity = 5)
    @capacity = capacity
    @lock     = Mutex.new
    @empty    = ConditionVariable.new
    @full     = ConditionVariable.new
    @queue    = []
  end

  def push(o)
    @lock.synchronize do
      while queue.length >= capacity
        @empty.wait(@lock)
      end
      queue.push(o)
      @full.signal
    end
  end

  def pop
    @lock.synchronize do
      while queue.length == 0
        @full.wait(@lock)
      end
      o = queue.shift
      @empty.signal
      o
    end
  end
end

buffer = ArrayBlockingQueue.new(3)
producer = Thread.new do
  i = 0
  while true
    buffer.push(i)
    i += 1
    s = rand(10) / 2
    puts "Producer sleep #{s}"
    sleep(s)
  end
end

consumer = Thread.new do
  while true
    buffer.pop
    s = rand(10)
    puts "\t\t\tConsumer sleep #{s}"
    sleep(s)
  end
end

producer.join
consumer.join