Simple broadcast using ruby-concurrent

concurrent_broadcast.rb

require 'concurrent-edge'

class Broadcast
  def initialize &block
    @channel = {}
    @listener = {}
    instance_exec(&block)
    Concurrent::Channel.go { start_consumer  }
  end

  def on event, &block
    @channel[event] = Concurrent::Channel.new
    @listener[event] = block
  end

  def publish event, *data
    @channel[event] << data
  end

  def start_consumer
    @channel.each do |event, channel|
      Thread.new(event, channel) do |evt,ch|
        while true
          if data = ~ch
            @listener[evt].call(*data)
          end
        end
      end
    end
  end
end

b = Broadcast.new do
  on 'ping' do |data|
    publish 'pong', Time.now, data
  end
  on 'pong' do |date, i|
    puts "#{i} Ponged in #{Time.now - date}"
  end
end


b.publish 'ping'