Class: AMQP::Queue

ch       = AMQP::Channel.new(connection)
exchange = ch.direct('backlog.events')
queue    = ch.queue('', :exclusive => true)
queue.bind(exchange)

ch       = AMQP::Channel.new(connection)
queue    = ch.queue('', :exclusive => true)
queue.bind('backlog.events')

Parameters:

• Exchange (Exchange) —

  to bind to. May also be a string or any object that responds to #name.

• opts (Hash) (defaults to: {}) —

  a customizable set of options

Options Hash (opts):

• :routing_key (String) —

  Specifies the routing key for the binding. The routing key is used for routing messages depending on the exchange configuration. Not all exchanges use a routing key! Refer to the specific exchange documentation. If the routing key is empty and the queue name is empty, the routing key will be the current queue for the channel, which is the last declared queue.

• :arguments (Hash) — default: nil —

  A hash of optional arguments with the declaration. Headers exchange type uses these metadata attributes for routing matching. In addition, brokers may implement AMQP extensions using x-prefixed declaration arguments.

• :nowait (Boolean) — default: true —

  If set, the server will not respond to the method. The client should not wait for a reply method. If the server could not complete the method it will raise a channel or connection exception.

Yields:

• [] Since queue.bind-ok carries no attributes, no parameters are yielded to the block.

Returns:

• (Queue) —

  Self

See Also:

• #unbind

Returns:

• (String) —

  Consumer tag of the default consumer associated with this queue (if any), or nil

See Also:

• #subscribe
• Consumer

Returns:

• (AMQP::Consumer) —

  Default consumer associated with this queue (if any), or nil

See Also:

• #subscribe
• Consumer

Parameters:

• opts (Hash) (defaults to: {}) —

  a customizable set of options

Options Hash (opts):

• :if_unused (Boolean) — default: false —

  If set, the server will only delete the queue if it has no consumers. If the queue has consumers the server does does not delete it but raises a channel exception instead.

• :if_empty (Boolean) — default: false —

  If set, the server will only delete the queue if it has no messages. If the queue is not empty the server raises a channel exception.

• :nowait (Boolean) — default: false —

  If set, the server will not respond to the method. The client should not wait for a reply method. If the server could not complete the method it will raise a channel or connection exception.

Yields:

• (delete_ok) —

  Yields AMQP method (queue.delete-ok) instance.

Yield Parameters:

• delete_ok (AMQP::Protocol::Queue::DeleteOk) —

  AMQP queue.delete-ok) instance. Carries number of messages that were in the queue.

Returns:

• (NilClass) —

  nil (for v0.7 compatibility)

• (NilClass) —

  nil (for v0.7 compatibility)

See Also:

• #purge
• #unbind

queue.pop do |metadata, payload|
  if payload
    puts "Fetched a message: #{payload.inspect}, content_type: #{metadata.content_type}. Shutting down..."
  else
    puts "No messages in the queue"
  end
end

Parameters:

• opts (Hash) (defaults to: {}) —

  a customizable set of options

Options Hash (opts):

• :ack (Boolean) — default: false —

  If this field is set to false the server does not expect acknowledgments for messages. That is, when a message is delivered to the client the server automatically and silently acknowledges it on behalf of the client. This functionality increases performance but at the cost of reliability. Messages can get lost if a client dies before it can deliver them to the application.

Yields:

• (headers, payload) —

  When block only takes one argument, yields payload to it. In case of two arguments, yields headers and payload.

Yield Parameters:

• headers (AMQP::Header) —

  Headers (metadata) associated with this message (for example, routing key).

• payload (String) —

  Message body (content). On Ruby 1.9, you may want to check or enforce content encoding.

Returns:

• (Qeueue) —

  Self

Parameters:

• opts (Hash) (defaults to: {}) —

  a customizable set of options

Options Hash (opts):

• :nowait (Boolean) — default: false —

  If set, the server will not respond to the method. The client should not wait for a reply method. If the server could not complete the method it will raise a channel or connection exception.

Yields:

• (purge_ok) —

  Yields AMQP method (queue.purge-ok) instance.

Yield Parameters:

• purge_ok (AMQP::Protocol::Queue::PurgeOk) —

  AMQP queue.purge-ok) instance. Carries number of messages that were purged.

Returns:

• (NilClass) —

  nil (for v0.7 compatibility)

See Also:

• #delete
• #unbind

Returns:

• (Boolean) —

  true if this queue is server-named

AMQP::Channel.queue('name').status { |number_of_messages, number_of_active_consumers|
  puts number_of_messages
}

Yields:

• (number_of_messages, number_of_active_consumers)

Yield Parameters:

• number_of_messages (Fixnum) —

  Number of messages in the queue

• number_of_active_consumers (Fixnum) —

  Number of active consumers for the queue. Note that consumers can suspend activity (Channel.Flow) in which case they do not appear in this count.

Raises:

• (ArgumentError)

EventMachine.run do
  exchange = AMQP::Channel.direct("foo queue")
  EM.add_periodic_timer(1) do
    exchange.publish("random number #{rand(1000)}")
  end

  queue = AMQP::Channel.queue('foo queue')
  queue.subscribe { |body| puts "received payload [#{body}]" }
end

EventMachine.run do
  connection = AMQP.connect(:host => '127.0.0.1')
  puts "Connected to AMQP broker. Running #{AMQP::VERSION} version of the gem..."

  channel  = AMQP::Channel.new(connection)
  queue    = channel.queue("amqpgem.examples.hello_world", :auto_delete => true)
  exchange = channel.direct("amq.direct")

  queue.bind(exchange)

  channel.on_error do |ch, channel_close|
    puts channel_close.reply_text
    connection.close { EventMachine.stop }
  end

  queue.subscribe do |metadata, payload|
    puts "metadata.routing_key : #{metadata.routing_key}"
    puts "metadata.content_type: #{metadata.content_type}"
    puts "metadata.priority    : #{metadata.priority}"
    puts "metadata.headers     : #{metadata.headers.inspect}"
    puts "metadata.timestamp   : #{metadata.timestamp.inspect}"
    puts "metadata.type        : #{metadata.type}"
    puts "metadata.delivery_tag: #{metadata.delivery_tag}"
    puts "metadata.redelivered : #{metadata.redelivered}"

    puts "metadata.app_id      : #{metadata.app_id}"
    puts "metadata.exchange    : #{metadata.exchange}"
    puts
    puts "Received a message: #{payload}. Disconnecting..."

    connection.close {
      EventMachine.stop { exit }
    }
  end

  exchange.publish("Hello, world!",
                   :app_id      => "amqpgem.example",
                   :priority    => 8,
                   :type        => "kinda.checkin",
                   # headers table keys can be anything
                   :headers     => {
                     :coordinates => {
                       :latitude  => 59.35,
                       :longitude => 18.066667
                     },
                     :participants => 11,
                     :venue        => "Stockholm"
                   },
                   :timestamp   => Time.now.to_i)
end

class Consumer

  #
  # API
  #

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

    @channel    = channel
    # Consumer#handle_channel_exception will handle channel
    # exceptions. Keep in mind that you can only register one error handler,
    # so the last one registered "wins".
    @channel.on_error(&method(:handle_channel_exception))
  end # initialize

  def start
    @queue = @channel.queue(@queue_name, :exclusive => true)
    # #handle_message method will be handling messages routed to @queue
    @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 Consumer

  #
  # API
  #

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

class Worker

  #
  # API
  #

  def initialize(channel, queue_name = AMQ::Protocol::EMPTY_STRING, consumer = Consumer.new)
    @queue_name = queue_name

    @channel    = channel
    @channel.on_error(&method(:handle_channel_exception))

    @consumer   = consumer
  end # initialize

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

  #
  # Implementation
  #

  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

require "ostruct"
require "json"

# RSpec example
describe Consumer do
  describe "when a new message arrives" do
    subject { described_class.new }

    let(:metadata) do
      o = OpenStruct.new

      o.content_type = "application/json"
      o
    end
    let(:payload)  { JSON.encode({ :command => "reload_config" }) }

    it "does some useful work" do
      # check preconditions here if necessary

      subject.handle_message(metadata, payload)

      # add your code expectations here
    end
  end
end

Parameters:

• opts (Hash) (defaults to: {}) —

  a customizable set of options

Options Hash (opts):

• :ack (Boolean) — default: false —

  If this field is set to false the server does not expect acknowledgments for messages. That is, when a message is delivered to the client the server automatically and silently acknowledges it on behalf of the client. This functionality increases performance but at the cost of reliability. Messages can get lost if a client dies before it can deliver them to the application.

• :nowait (Boolean) — default: false —

  If set, the server will not respond to the method. The client should not wait for a reply method. If the server could not complete the method it will raise a channel or connection exception.

• :confirm (#call) — default: nil —

  If set, this proc will be called when the server confirms subscription to the queue with a basic.consume-ok message. Setting this option will automatically set :nowait => false. This is required for the server to send a confirmation.

• :exclusive (Boolean) — default: false —

  Request exclusive consumer access, meaning only this consumer can access the queue. This is useful when you want a long-lived shared queue to be temporarily accessible by just one application (or thread, or process). If application exclusive consumer is part of crashes or loses network connection to the broker, channel is closed and exclusive consumer is thus cancelled.

Yields:

• (headers, payload) —

  When block only takes one argument, yields payload to it. In case of two arguments, yields headers and payload.

Yield Parameters:

• headers (AMQP::Header) —

  Headers (metadata) associated with this message (for example, routing key).

• payload (String) —

  Message body (content). On Ruby 1.9, you may want to check or enforce content encoding.

Returns:

• (Queue) —

  Self

Raises:

• (RuntimeError)

See Also:

• Documentation guide on queues
• #unsubscribe
• Consumer

Returns:

• (Boolean) —

  true if there is a consumer tag associated with this Queue instance

Parameters:

• Exchange (Exchange) —

  to unbind from.

• opts (Hash) (defaults to: {}) —

  a customizable set of options

Options Hash (opts):

• :routing_key (String) —

  Binding routing key

• :arguments (Hash) —

  Binding arguments

• :nowait (Boolean) — default: true —

  If set, the server will not respond to the method. The client should not wait for a reply method. If the server could not complete the method it will raise a channel or connection exception.

Yields:

• [] Since queue.unbind-ok carries no attributes, no parameters are yielded to the block.

See Also:

• #bind

Parameters:

• opts (Hash) (defaults to: {}) —

  a customizable set of options

Options Hash (opts):

• :nowait (Boolean) — default: true —

  If set, the server will not respond to the method. The client should not wait for a reply method, the callback (if passed) will be ignored. If the server could not complete the method it will raise a channel or connection exception.

Yields:

• (cancel_ok)

Yield Parameters:

• cancel_ok (AMQP::Protocol::Basic::CancelOk) —

  AMQP method basic.cancel-ok. You can obtain consumer tag from it.