stuff.rb

circular_buffer.rb

class CircularBuffer
  BUFFER_SIZE = 10

  def initialize
    @buffer_array = Array.new(BUFFER_SIZE)
    @count = 0
    @out_counter = 0
    @in_counter = 0
  end

  def <<(val)
    @buffer_array[@in_counter % BUFFER_SIZE] = val
    @count += 1
    @in_counter += 1
    @buffer_array
  end

  def pop
    idx = @out_counter % BUFFER_SIZE
    val = @buffer_array[idx]
    @buffer_array[idx] = nil
    @count -= 1
    @out_counter += 1
    val
  end

  def any?
    @count > 0
  end

  def open?
    @count < BUFFER_SIZE
  end

  def to_s
    @buffer_array.inspect
  end
end

consumer.rb

class Consumer < QueueClient
  def initialize(&producer_function)
    @consumer_function = producer_function
    super
  end

  def perform
    loop do
      break if buffer.dead? && buffer.empty? 
      #sleep(rand(0.01..0.5)) #simulate real work being done
      @consumer_function.call(buffer.pop)
      @buffer_value_counter += 1
    end
  end
end

main.rb

require "./circular_buffer.rb"
require "./queue.rb"
require "./queue_client.rb"
require "./producer.rb"
require "./consumer.rb"
require 'pry'

count = 0

buffer = SimpleQueue.new

producer = Producer.new do
  "Hello #{count += 1}"
end

consumer = Consumer.new do |e|
  puts e
end

buffer.register(producer, consumer)
p_thread = producer.run
c_thread = consumer.run
c_thread.join
p_thread.join

puts "Consumed #{consumer.buffer_value_counter}"
puts "Produced #{producer.buffer_value_counter}"

producer.rb

class Producer < QueueClient
  def initialize(&producer_function)
    @producer_function = producer_function
    super
  end

  def perform
    1000.times do
      #sleep(rand(0.01..0.5)) #simulate real work being done. 
      buffer << @producer_function.call
      @buffer_value_counter += 1
    end
  end
end

queue.rb

class SimpleQueue
  attr_accessor :buffer, :client_list

  def initialize
    @buffer = CircularBuffer.new
    @client_list = {producers: [], consumers: []}
  end

  def <<(val)
    unless buffer.open?
      Thread.stop
    end
    @buffer << val
    notify_clients :consumers
    @buffer
  end


  def pop
    unless buffer.any?
      Thread.stop
    end
    val = @buffer.pop
    notify_clients :producers
    val
  end
  
  def to_s
    @buffer.to_s
  end

  def register(*buffer_clients)
    buffer_clients.each do |x|
      case x
      when Producer
        @client_list[:producers] << x
      when Consumer
        @client_list[:consumers] << x
      end
      x.buffer = self
    end
  end

  def dead?
    @client_list[:producers].all?{|x|x.dead?}
  end

  def empty?
    [email protected]?
  end

  private

  def notify_clients(client_type)
    @client_list[client_type].each(&:notify_thread)
  end
end

queue_client.rb

class QueueClient
  attr_accessor :buffer, :thread, :buffer_value_counter

  def initialize
    @buffer_value_counter = 0
  end

  def run
    raise StandardError.new("Must be registered with a buffer") if buffer.nil?
    @thread = Thread.new { perform }
  end

  def dead?
    !self.thread.alive?
  end

  def notify_thread
    self.thread.run if self.thread.status == 'sleep'
  end
end