RabbitMQ Topic for Pub/Sub

demo.log

% ruby sub_update.rb
# output after ruby publisher.rb
Update form: UPDATE 89, routing key is data.update

% ruby sub_logger.rb
# output after ruby publisher.rb
Log: UPDATE 89, routing key is data.update
Log: CREATE 90, routing key is data.create

% ruby publisher.rb

publisher.rb

#!/usr/bin/env ruby
# encoding: utf-8

require "rubygems"
require "amqp"

EXCHANGE = "data"

AMQP.start("amqp://127.0.0.1") do |connection|
  channel = AMQP::Channel.new(connection)
  exchange = channel.topic(EXCHANGE)

  # one create and one update
  exchange.publish("UPDATE 89", :routing_key => "data.update")
  exchange.publish("CREATE 90", :routing_key => "data.create")

  show_stopper = Proc.new {
    connection.close { EventMachine.stop }
  }
  EM.add_timer(2, show_stopper)
end

sub_logger.rb

#!/usr/bin/env ruby
# encoding: utf-8

require "rubygems"
require "amqp"

EXCHANGE = "data"
QUEUE = "data.log"

EventMachine.run do
  AMQP.connect do |connection|
    channel = AMQP::Channel.new(connection)
    exchange = channel.topic(EXCHANGE)

    # Subscribe to all data events to log them
    binding = channel.queue(QUEUE, :exclusive => true).bind(exchange, :routing_key => "data.#")
    binding.subscribe do |headers, payload|
      puts "Log: #{payload}, routing key is #{headers.routing_key}"
    end
  end
end

sub_udpate.rb

#!/usr/bin/env ruby
# encoding: utf-8

require "rubygems"
require "amqp"

EXCHANGE = "data"
QUEUE = "data.update"

EventMachine.run do
  AMQP.connect do |connection|
    channel = AMQP::Channel.new(connection)
    exchange = channel.topic(EXCHANGE)

    # Only subscribe to data.update
    binding = channel.queue(QUEUE).bind(exchange, :routing_key => "data.update")
    binding.subscribe do |headers, payload|
      puts "Update form: #{payload}, routing key is #{headers.routing_key}"
    end
  end
end