amqp-demo3.rb

require "rubygems"
require "eventmachine"
require "mq"
require "amqp"
require "bunny"

#PUBLISH_MODE = :one_to_n
PUBLISH_MODE = :one_to_one_of_n

START_TIME = Time.new

def finish
  EM.forks.each {|pid| Process.kill("KILL", pid)}
end

def gen_qname
  if PUBLISH_MODE == :one_to_n
    rand(2 ** 16).to_s(16)
  else
    "foobarbaz"
  end
end

def timestamp
  delta_t = Time.new - START_TIME
  delta_t.to_s
end

EM.fork(3) do
  mq = MQ.new
  qname = gen_qname
  puts "starting queue ``#{qname}'', PID: #{Process.pid.to_s}"
  q = mq.queue(qname).bind(mq.topic("test"), :key => "test.key")
  q.subscribe do |msg|
    puts "[queue #{qname} (pid #{Process.pid.to_s})] received msg: #{msg} @ #{timestamp}"
  end
end

connection = Bunny.new
connection.start

topic = connection.exchange("test", :type => :topic)
(1..10).each do |i|
  topic.publish("[#{i}] " + timestamp, :key => "test.key")
  sleep(0.1)
end
    
finish
connection.stop