Simple EM rowly and parellel chains

em_chain.rb

class EMChain
  attr_accessor :queue, :complete_clb, :item_block
  attr_accessor :in_progress_queue
  attr_accessor :max
  def initialize(&config_blk)
    self.queue = []
    self.in_progress_queue = {}    
    config_blk.call self if config_blk
  end
  
  
  def try_next
    while queue.size > 0 && (!@max || in_progress_queue.size < @max)
      run_next_block
    end
  end
  
  def parallel(max = nil)
    @max = max
    start do
      try_next
    end
    self
  end
  
  def rowly
    parallel 1
  end
  
  def max=(v)
    @max = v
    try_next
    v
  end
  
  def block(&clb)
    queue << clb
    try_next if @started
    clb
  end
  
  def on_complete(&clb)
    self.complete_clb = clb
  end
  
  private
  
  def start
    @started = true
    if queue.empty?
      complete
    else
      yield
    end
  end
  
  def complete
    complete_clb.call if complete_clb
  end
  
  def run_next_block 
    if (block = queue.shift)
      in_progress_queue[block] = true
      block.call ping_block(block)
    end
  end
  
  def ping_block(block)
    proc do
      in_progress_queue.delete block
      if in_progress_queue.size == 0 && queue.size == 0
        complete
      elsif queue.size != 0          
        run_next_block
      end
    end
  end
end

if __FILE__ == $0
  require 'rubygems'
  require 'pp'
  require 'eventmachine'

  
  def make_chain
    EMChain.new do |c|
      [:a, :b, :c].each do |item|
        c.block do |ping_block|
          puts "Start #{item}"
          EM.add_timer(1) do
            puts "Run #{item}"
            ping_block.call
          end
        end
      end
      c.on_complete do
        puts "Completed!"
        EM.stop
      end
    end
  end
  chain = make_chain
  puts "Rowly start"
  EM.run do
    chain.rowly
  end

  # =>
  # Rowly start
  # Start a
  # Run a
  # Start b
  # Run b
  # Start c
  # Run c
  # Completed!


  chain = make_chain

  puts "Parallel start"
  EM.run do
    chain.parallel
  end

  # =>
  # Parallel start
  # Start a
  # Start b
  # Start c
  # Run a
  # Run b
  # Run c
  # Completed!

  chain = make_chain

  puts "Parallel 2 start"
  EM.run do
    chain.parallel(2)
    chain.block do |ping_block|
      puts "Start d"
      EM.add_timer(1) do
        puts "Run d"
        ping_block.call
      end
    end
  end

  # =>
  # Parallel 2 start
  # Start a
  # Start b
  # Run a
  # Start c
  # Run b
  # Start d
  # Run c
  # Run d
  # Completed!
end