Contents

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

require "rubygems"
require "amqp"

class Consumer

  #
  # API
  #

  def initialize(channel, queue_name = AMQ::Protocol::EMPTY_STRING)
    @queue_name = queue_name

    @channel    = channel
    @channel.on_error(&method(:handle_channel_exception))
  end # initialize

  def start
    @queue = @channel.queue(@queue_name, :exclusive => true)
    @queue.subscribe(&method(:handle_message))
  end # start



  #
  # Implementation
  #

  def handle_message(metadata, payload)
    puts "Received a message: #{payload}, content_type = #{metadata.content_type}"
  end # handle_message(metadata, payload)

  def handle_channel_exception(channel, channel_close)
    puts "Oops... a channel-level exception: code = #{channel_close.reply_code}, message = #{channel_close.reply_text}"
  end # handle_channel_exception(channel, channel_close)
end


class Producer

  #
  # API
  #

  def initialize(channel, exchange)
    @channel  = channel
    @exchange = exchange
  end # initialize(channel, exchange)

  def publish(message, options = {})
    @exchange.publish(message, options)
  end # publish(message, options = {})
end


AMQP.start("amqp://guest:[e-mail]") do |connection, open_ok|
  channel  = AMQP::Channel.new(connection)
  worker   = Consumer.new(channel, "amqpgem.objects.integration")
  worker.start

  producer = Producer.new(channel, channel.default_exchange)
  puts "Publishing..."
  producer.publish("Hello, world", :routing_key => "amqpgem.objects.integration")

  # stop in 2 seconds
  EventMachine.add_timer(2.0) { connection.close { EventMachine.stop } }
end

Version data entries

53 entries across 53 versions & 1 rubygems

Version	Path
amqp-0.9.2	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.9.1	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.9.0	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.9.0.pre3	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.9.0.pre2	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.9.0.pre1	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.8.4	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.8.3	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.8.2	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.8.1	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.8.0	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.8.0.rc15	examples/guides/queues/12_objects_that_consume_messages.rb
amqp-0.8.0.rc14	examples/guides/queues/12_objects_that_consume_messages.rb