class Bunny::Channel

## Channels in RabbitMQ

To quote {www.rabbitmq.com/resources/specs/amqp0-9-1.pdf AMQP 0.9.1 specification}:

AMQP 0.9.1 is a multi-channelled protocol. Channels provide a way to multiplex a heavyweight TCP/IP connection into several light weight connections. This makes the protocol more “firewall friendly” since port usage is predictable. It also means that traffic shaping and other network QoS features can be easily employed. Channels are independent of each other and can perform different functions simultaneously with other channels, the available bandwidth being shared between the concurrent activities.

## Opening Channels

Channels can be opened either via `Bunny::Session#create_channel` (sufficient in the majority of cases) or by instantiating `Bunny::Channel` directly:

conn = Bunny.new
conn.start

ch   = conn.create_channel

This will automatically allocate a channel id.

## Closing Channels

Channels are closed via {Bunny::Channel#close}. Channels that get a channel-level exception are closed, too. Closed channels can no longer be used. Attempts to use them will raise {Bunny::ChannelAlreadyClosed}.

ch = conn.create_channel
ch.close

## Higher-level API

Bunny offers two sets of methods on {Bunny::Channel}: known as higher-level and lower-level APIs, respectively. Higher-level API mimics {rubyamqp.info amqp gem} API where exchanges and queues are objects (instance of {Bunny::Exchange} and {Bunny::Queue}, respectively). Lower-level API is built around AMQP 0.9.1 methods (commands), where queues and exchanges are passed as strings (à la RabbitMQ Java client, {clojurerabbitmq.info Langohr} and Pika).

### Queue Operations In Higher-level API

### Exchange Operations In Higher-level API

## Channel Qos (Prefetch Level)

It is possible to control how many messages at most a consumer will be given (before it acknowledges or rejects previously consumed ones). This setting is per channel and controlled via {Bunny::Channel#prefetch}.

## Channel IDs

Channels are identified by their ids which are integers. Bunny takes care of allocating and releasing them as channels are opened and closed. It is almost never necessary to specify channel ids explicitly.

There is a limit on the maximum number of channels per connection, usually 65536. Note that allocating channels is very cheap on both client and server so having tens, hundreds or even thousands of channels is not a problem.

## Channels and Error Handling

Channel-level exceptions are more common than connection-level ones and often indicate issues applications can recover from (such as consuming from or trying to delete a queue that does not exist).

With Bunny, channel-level exceptions are raised as Ruby exceptions, for example, {Bunny::NotFound}, that provide access to the underlying `channel.close` method information.

@example Handling 404 NOT_FOUND

begin
  ch.queue_delete("queue_that_should_not_exist#{rand}")
rescue Bunny::NotFound => e
  puts "Channel-level exception! Code: #{e.channel_close.reply_code}, message: #{e.channel_close.reply_text}"
end

@example Handling 406 PRECONDITION_FAILED

begin
  ch2 = conn.create_channel
  q   = "bunny.examples.recovery.q#{rand}"

  ch2.queue_declare(q, :durable => false)
  ch2.queue_declare(q, :durable => true)
rescue Bunny::PreconditionFailed => e
  puts "Channel-level exception! Code: #{e.channel_close.reply_code}, message: #{e.channel_close.reply_text}"
ensure
  conn.create_channel.queue_delete(q)
end

@see www.rabbitmq.com/tutorials/amqp-concepts.html AMQP 0.9.1 Model Concepts Guide @see rubybunny.info/articles/getting_started.html Getting Started with RabbitMQ Using Bunny @see rubybunny.info/articles/queues.html Queues and Consumers @see rubybunny.info/articles/exchanges.html Exchanges and Publishing @see rubybunny.info/articles/error_handling.html Error Handling and Recovery Guide

Constants

DEFAULT_CONTENT_TYPE
MAX_PREFETCH_COUNT

prefetch_count is of type short in the protocol. MK.

SHORTSTR_LIMIT

Attributes

connection[R]

@return [Bunny::Session] AMQP connection this channel was opened on

consumers[R]

@return [Hash<String, Bunny::Consumer>] Consumer instances declared on this channel

exchanges[R]

@return [Hash<String, Bunny::Exchange>] Exchange instances declared on this channel

id[RW]

@return [Integer] Channel id

nacked_set[R]

@return [Set<Integer>] Set of nacked message indexes that have been nacked

next_publish_seq_no[R]

@return [Integer] Next publisher confirmations sequence index

prefetch_count[R]

@return [Integer] active basic.qos prefetch value

prefetch_global[R]

@return [Integer] active basic.qos prefetch global mode

queues[R]

@return [Hash<String, Bunny::Queue>] Queue instances declared on this channel

recoveries_counter[R]
status[R]

@return [Symbol] Channel status (:opening, :open, :closed)

unconfirmed_set[R]

@return [Set<Integer>] Set of published message indexes that are currently unconfirmed

work_pool[R]

@return [Bunny::ConsumerWorkPool] Thread pool delivered messages are dispatched to.

Public Class Methods

new(connection = nil, id = nil, work_pool = ConsumerWorkPool.new(1)) click to toggle source

@param [Bunny::Session] connection AMQP 0.9.1 connection @param [Integer] id Channel id, pass nil to make Bunny automatically allocate it @param [Bunny::ConsumerWorkPool] work_pool Thread pool for delivery processing, by default of size 1

# File lib/bunny/channel.rb, line 167
def initialize(connection = nil, id = nil, work_pool = ConsumerWorkPool.new(1))
  @connection = connection
  @logger     = connection.logger
  @id         = id || @connection.next_channel_id

  # channel allocator is exhausted
  if @id < 0
    msg = "Cannot open a channel: max number of channels on connection reached. Connection channel_max value: #{@connection.channel_max}"
    @logger.error(msg)

    raise msg
  else
    @logger.debug { "Allocated channel id: #{@id}" }
  end

  @status     = :opening

  @connection.register_channel(self)

  @queues     = Hash.new
  @exchanges  = Hash.new
  @consumers  = Hash.new
  @work_pool  = work_pool

  # synchronizes frameset delivery. MK.
  @publishing_mutex = @connection.mutex_impl.new
  @consumer_mutex   = @connection.mutex_impl.new

  @queue_mutex    = @connection.mutex_impl.new
  @exchange_mutex = @connection.mutex_impl.new

  @unconfirmed_set_mutex = @connection.mutex_impl.new

  self.reset_continuations

  # threads awaiting on continuations. Used to unblock
  # them when network connection goes down so that busy loops
  # that perform synchronous operations can work. MK.
  @threads_waiting_on_continuations           = Set.new
  @threads_waiting_on_confirms_continuations  = Set.new
  @threads_waiting_on_basic_get_continuations = Set.new

  @next_publish_seq_no = 0
  @delivery_tag_offset = 0

  @recoveries_counter = Bunny::Concurrent::AtomicFixnum.new(0)
  @uncaught_exception_handler = Proc.new do |e, consumer|
    @logger.error "Uncaught exception from consumer #{consumer.to_s}: #{e.inspect} @ #{e.backtrace[0]}"
  end
end

Public Instance Methods

ack(delivery_tag, multiple = false) click to toggle source

Acknowledges a message. Acknowledged messages are completely removed from the queue.

@param [Integer] delivery_tag Delivery tag to acknowledge @param [Boolean] multiple (false) Should all unacknowledged messages up to this be acknowledged as well? @see Bunny::Channel#nack @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 481
def ack(delivery_tag, multiple = false)
  basic_ack(delivery_tag.to_i, multiple)
end
Also aliased as: acknowledge
acknowledge(delivery_tag, multiple = false)
Alias for: ack
active() click to toggle source

@return [Boolean] true if this channel is open

# File lib/bunny/channel.rb, line 277
def active
  open?
end
add_consumer(queue, consumer_tag, no_ack, exclusive, arguments, &block) click to toggle source

@private

# File lib/bunny/channel.rb, line 1633
def add_consumer(queue, consumer_tag, no_ack, exclusive, arguments, &block)
  @consumer_mutex.synchronize do
    c = Consumer.new(self, queue, consumer_tag, no_ack, exclusive, arguments)
    c.on_delivery(&block) if block
    @consumers[consumer_tag] = c
  end
end
any_consumers?() click to toggle source

@return [Boolean] true if there are consumers on this channel @api public

# File lib/bunny/channel.rb, line 978
def any_consumers?
  @consumer_mutex.synchronize { @consumers.any? }
end
basic_ack(delivery_tag, multiple = false) click to toggle source

Acknowledges a delivery (message).

@param [Integer] delivery_tag Delivery tag obtained from delivery info @param [Boolean] multiple Should all deliveries up to this one be acknowledged? @return [NilClass] nil

@example Ack a message

conn  = Bunny.new
conn.start

ch    = conn.create_channel
q.subscribe do |delivery_info, properties, payload|
  # requeue the message
  ch.basic_ack(delivery_info.delivery_tag.to_i)
end

@example Ack a message fetched via basic.get

conn  = Bunny.new
conn.start

ch    = conn.create_channel
# we assume the queue exists and has messages
delivery_info, properties, payload = ch.basic_get("bunny.examples.queue3", :manual_ack => true)
ch.basic_ack(delivery_info.delivery_tag.to_i)

@example Ack multiple messages fetched via basic.get

conn  = Bunny.new
conn.start

ch    = conn.create_channel
# we assume the queue exists and has messages
_, _, payload1 = ch.basic_get("bunny.examples.queue3", :manual_ack => true)
_, _, payload2 = ch.basic_get("bunny.examples.queue3", :manual_ack => true)
delivery_info, properties, payload3 = ch.basic_get("bunny.examples.queue3", :manual_ack => true)
# ack all fetched messages up to payload3
ch.basic_ack(delivery_info.delivery_tag.to_i, true)

@see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 767
def basic_ack(delivery_tag, multiple = false)
  guarding_against_stale_delivery_tags(delivery_tag) do
    raise_if_no_longer_open!
    @connection.send_frame(AMQ::Protocol::Basic::Ack.encode(@id, delivery_tag, multiple))

    nil
  end
end
basic_cancel(consumer_tag) click to toggle source

Removes a consumer. Messages for this consumer will no longer be delivered. If the queue it was on is auto-deleted and this consumer was the last one, the queue will be deleted.

@param [String] consumer_tag Consumer tag (unique identifier) to cancel

@return [AMQ::Protocol::Basic::CancelOk] RabbitMQ response @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 962
def basic_cancel(consumer_tag)
  @connection.send_frame(AMQ::Protocol::Basic::Cancel.encode(@id, consumer_tag, false))

  with_continuation_timeout do
    @last_basic_cancel_ok = wait_on_continuations
  end

  # reduces thread usage for channels that don't have any
  # consumers
  @work_pool.shutdown(true) unless self.any_consumers?

  @last_basic_cancel_ok
end
basic_consume(queue, consumer_tag = generate_consumer_tag, no_ack = false, exclusive = false, arguments = nil, &block) click to toggle source

Registers a consumer for queue. Delivered messages will be handled with the block provided to this method.

@param [String, Bunny::Queue] queue Queue to consume from @param [String] consumer_tag Consumer tag (unique identifier), generated by Bunny by default @param [Boolean] no_ack (false) If true, delivered messages will be automatically acknowledged.

If false, manual acknowledgements will be necessary.

@param [Boolean] exclusive (false) Should this consumer be exclusive? @param [Hash] arguments (nil) Optional arguments that may be used by RabbitMQ extensions, etc

@return [AMQ::Protocol::Basic::ConsumeOk] RabbitMQ response @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 854
def basic_consume(queue, consumer_tag = generate_consumer_tag, no_ack = false, exclusive = false, arguments = nil, &block)
  raise_if_no_longer_open!
  maybe_start_consumer_work_pool!

  queue_name = if queue.respond_to?(:name)
                 queue.name
               else
                 queue
               end

  # helps avoid race condition between basic.consume-ok and basic.deliver if there are messages
  # in the queue already. MK.
  if consumer_tag && consumer_tag.strip != AMQ::Protocol::EMPTY_STRING
    add_consumer(queue_name, consumer_tag, no_ack, exclusive, arguments, &block)
  end

  @connection.send_frame(AMQ::Protocol::Basic::Consume.encode(@id,
      queue_name,
      consumer_tag,
      false,
      no_ack,
      exclusive,
      false,
      arguments))

  begin
    with_continuation_timeout do
      @last_basic_consume_ok = wait_on_continuations
    end
  rescue Exception => e
    # if basic.consume-ok never arrives, unregister the proactively
    # registered consumer. MK.
    unregister_consumer(@last_basic_consume_ok.consumer_tag)

    raise e
  end

  # in case there is another exclusive consumer and we get a channel.close
  # response here. MK.
  raise_if_channel_close!(@last_basic_consume_ok)

  # covers server-generated consumer tags
  add_consumer(queue_name, @last_basic_consume_ok.consumer_tag, no_ack, exclusive, arguments, &block)

  @last_basic_consume_ok
end
Also aliased as: consume
basic_consume_with(consumer) click to toggle source

Registers a consumer for queue as {Bunny::Consumer} instance.

@param [Bunny::Consumer] consumer Consumer to register. It should already have queue name, consumer tag

and other attributes set.

@return [AMQ::Protocol::Basic::ConsumeOk] RabbitMQ response @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 910
def basic_consume_with(consumer)
  raise_if_no_longer_open!
  maybe_start_consumer_work_pool!

  # helps avoid race condition between basic.consume-ok and basic.deliver if there are messages
  # in the queue already. MK.
  if consumer.consumer_tag && consumer.consumer_tag.strip != AMQ::Protocol::EMPTY_STRING
    register_consumer(consumer.consumer_tag, consumer)
  end

  @connection.send_frame(AMQ::Protocol::Basic::Consume.encode(@id,
      consumer.queue_name,
      consumer.consumer_tag,
      false,
      consumer.no_ack,
      consumer.exclusive,
      false,
      consumer.arguments))

  begin
    with_continuation_timeout do
      @last_basic_consume_ok = wait_on_continuations
    end
  rescue Exception => e
    # if basic.consume-ok never arrives, unregister the proactively
    # registered consumer. MK.
    unregister_consumer(@last_basic_consume_ok.consumer_tag)

    raise e
  end

  # in case there is another exclusive consumer and we get a channel.close
  # response here. MK.
  raise_if_channel_close!(@last_basic_consume_ok)

  # covers server-generated consumer tags
  register_consumer(@last_basic_consume_ok.consumer_tag, consumer)

  raise_if_continuation_resulted_in_a_channel_error!

  @last_basic_consume_ok
end
Also aliased as: consume_with
basic_get(queue, opts = {:manual_ack => true}) click to toggle source

Synchronously fetches a message from the queue, if there are any. This method is for cases when the convenience of synchronous operations is more important than throughput.

@param [String] queue Queue name @param [Hash] opts Options

@option opts [Boolean] :ack (true) [DEPRECATED] Use :manual_ack instead @option opts [Boolean] :manual_ack (true) Will this message be acknowledged manually?

@return [Array] A triple of delivery info, message properties and message content

@example Using Bunny::Channel#basic_get with manual acknowledgements

conn = Bunny.new
conn.start
ch   = conn.create_channel
# here we assume the queue already exists and has messages
delivery_info, properties, payload = ch.basic_get("bunny.examples.queue1", :manual_ack => true)
ch.acknowledge(delivery_info.delivery_tag)

@see Bunny::Queue#pop @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 594
def basic_get(queue, opts = {:manual_ack => true})
  raise_if_no_longer_open!

  unless opts[:ack].nil?
    warn "[DEPRECATION] `:ack` is deprecated.  Please use `:manual_ack` instead."
    opts[:manual_ack] = opts[:ack]
  end

  @connection.send_frame(AMQ::Protocol::Basic::Get.encode(@id, queue, !(opts[:manual_ack])))
  # this is a workaround for the edge case when basic_get is called in a tight loop
  # and network goes down we need to perform recovery. The problem is, basic_get will
  # keep blocking the thread that calls it without clear way to constantly unblock it
  # from the network activity loop (where recovery happens) with the current continuations
  # implementation (and even more correct and convenient ones, such as wait/notify, should
  # we implement them). So we return a triple of nils immediately which apps should be
  # able to handle anyway as "got no message, no need to act". MK.
  last_basic_get_response = if @connection.open?
                              begin
                                wait_on_basic_get_continuations
                              rescue Timeout::Error => e
                                raise_if_continuation_resulted_in_a_channel_error!
                                raise e
                              end
                            else
                              [nil, nil, nil]
                            end

  raise_if_continuation_resulted_in_a_channel_error!
  last_basic_get_response
end
basic_nack(delivery_tag, multiple = false, requeue = false) click to toggle source

Rejects or requeues messages just like {Bunny::Channel#basic_reject} but can do so with multiple messages at once.

@param [Integer] delivery_tag Delivery tag obtained from delivery info @param [Boolean] requeue Should the message be requeued? @param [Boolean] multiple Should all deliveries up to this one be rejected/requeued? @return [NilClass] nil

@example Requeue a message

conn  = Bunny.new
conn.start

ch    = conn.create_channel
q.subscribe do |delivery_info, properties, payload|
  # requeue the message
  ch.basic_nack(delivery_info.delivery_tag, false, true)
end

@example Reject a message

conn  = Bunny.new
conn.start

ch    = conn.create_channel
q.subscribe do |delivery_info, properties, payload|
  # requeue the message
  ch.basic_nack(delivery_info.delivery_tag)
end

@example Requeue a message fetched via basic.get

conn  = Bunny.new
conn.start

ch    = conn.create_channel
# we assume the queue exists and has messages
delivery_info, properties, payload = ch.basic_get("bunny.examples.queue3", :manual_ack => true)
ch.basic_nack(delivery_info.delivery_tag, false, true)

@example Requeue multiple messages fetched via basic.get

conn  = Bunny.new
conn.start

ch    = conn.create_channel
# we assume the queue exists and has messages
_, _, payload1 = ch.basic_get("bunny.examples.queue3", :manual_ack => true)
_, _, payload2 = ch.basic_get("bunny.examples.queue3", :manual_ack => true)
delivery_info, properties, payload3 = ch.basic_get("bunny.examples.queue3", :manual_ack => true)
# requeue all fetched messages up to payload3
ch.basic_nack(delivery_info.delivery_tag, true, true)

@see rubybunny.info/articles/queues.html Queues and Consumers guide @see rubybunny.info/articles/extensions.html RabbitMQ Extensions guide @api public

# File lib/bunny/channel.rb, line 829
def basic_nack(delivery_tag, multiple = false, requeue = false)
  guarding_against_stale_delivery_tags(delivery_tag) do
    raise_if_no_longer_open!
    @connection.send_frame(AMQ::Protocol::Basic::Nack.encode(@id,
                                                             delivery_tag,
                                                             multiple,
                                                             requeue))

    nil
  end
end
basic_publish(payload, exchange, routing_key, opts = {}) click to toggle source

Publishes a message using basic.publish AMQP 0.9.1 method.

@param [String] payload Message payload. It will never be modified by Bunny or RabbitMQ in any way. @param [String] exchange Exchange to publish to @param [String] routing_key Routing key @param [Hash] opts Publishing options

@option opts [Boolean] :persistent Should the message be persisted to disk? @option opts [Boolean] :mandatory Should the message be returned if it cannot be routed to any queue? @option opts [Integer] :timestamp A timestamp associated with this message @option opts [Integer] :expiration Expiration time after which the message will be deleted @option opts [String] :type Message type, e.g. what type of event or command this message represents. Can be any string @option opts [String] :reply_to Queue name other apps should send the response to @option opts [String] :content_type Message content type (e.g. application/json) @option opts [String] :content_encoding Message content encoding (e.g. gzip) @option opts [String] :correlation_id Message correlated to this one, e.g. what request this message is a reply for @option opts [Integer] :priority Message priority, 0 to 9. Not used by RabbitMQ, only applications @option opts [String] :message_id Any message identifier @option opts [String] :user_id Optional user ID. Verified by RabbitMQ against the actual connection username @option opts [String] :app_id Optional application ID

@return [Bunny::Channel] Self @api public

# File lib/bunny/channel.rb, line 532
def basic_publish(payload, exchange, routing_key, opts = {})
  raise_if_no_longer_open!
  raise ArgumentError, "routing key cannot be longer than #{SHORTSTR_LIMIT} characters" if routing_key && routing_key.size > SHORTSTR_LIMIT

  exchange_name = if exchange.respond_to?(:name)
                    exchange.name
                  else
                    exchange
                  end

  mode = if opts.fetch(:persistent, true)
           2
         else
           1
         end

  opts[:delivery_mode] ||= mode
  opts[:content_type]  ||= DEFAULT_CONTENT_TYPE
  opts[:priority]      ||= 0

  if @next_publish_seq_no > 0
    @unconfirmed_set_mutex.synchronize do
      @unconfirmed_set.add(@next_publish_seq_no)
      @next_publish_seq_no += 1
    end
  end

  frames = AMQ::Protocol::Basic::Publish.encode(@id,
    payload,
    opts,
    exchange_name,
    routing_key,
    opts[:mandatory],
    false,
    @connection.frame_max)
  @connection.send_frameset(frames, self)

  self
end
basic_qos(count, global = false) click to toggle source

Controls message delivery rate using basic.qos AMQP 0.9.1 method.

@param [Integer] prefetch_count How many messages can consumers on this channel be given at a time

(before they have to acknowledge or reject one of the earlier received messages)

@param [Boolean] global

Whether to use global mode for prefetch:
- +false+: per-consumer
- +true+:  per-channel
Note that the default value (+false+) hasn't actually changed, but
previous documentation described that as meaning per-channel and
unsupported in RabbitMQ, whereas it now actually appears to mean
per-consumer and supported
(https://www.rabbitmq.com/consumer-prefetch.html).

@return [AMQ::Protocol::Basic::QosOk] RabbitMQ response @see Bunny::Channel#prefetch @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 645
def basic_qos(count, global = false)
  raise ArgumentError.new("prefetch count must be a positive integer, given: #{count}") if count < 0
  raise ArgumentError.new("prefetch count must be no greater than #{MAX_PREFETCH_COUNT}, given: #{count}") if count > MAX_PREFETCH_COUNT
  raise_if_no_longer_open!

  @connection.send_frame(AMQ::Protocol::Basic::Qos.encode(@id, 0, count, global))

  with_continuation_timeout do
    @last_basic_qos_ok = wait_on_continuations
  end
  raise_if_continuation_resulted_in_a_channel_error!

  @prefetch_count  = count
  @prefetch_global = global

  @last_basic_qos_ok
end
Also aliased as: prefetch
basic_recover(requeue) click to toggle source

Redeliver unacknowledged messages

@param [Boolean] requeue Should messages be requeued? @return [AMQ::Protocol::Basic::RecoverOk] RabbitMQ response @api public

# File lib/bunny/channel.rb, line 669
def basic_recover(requeue)
  raise_if_no_longer_open!

  @connection.send_frame(AMQ::Protocol::Basic::Recover.encode(@id, requeue))
  with_continuation_timeout do
    @last_basic_recover_ok = wait_on_continuations
  end
  raise_if_continuation_resulted_in_a_channel_error!

  @last_basic_recover_ok
end
basic_reject(delivery_tag, requeue = false) click to toggle source

Rejects or requeues a message.

@param [Integer] delivery_tag Delivery tag obtained from delivery info @param [Boolean] requeue Should the message be requeued? @return [NilClass] nil

@example Requeue a message

conn  = Bunny.new
conn.start

ch    = conn.create_channel
q.subscribe do |delivery_info, properties, payload|
  # requeue the message
  ch.basic_reject(delivery_info.delivery_tag, true)
end

@example Reject a message

conn  = Bunny.new
conn.start

ch    = conn.create_channel
q.subscribe do |delivery_info, properties, payload|
  # reject the message
  ch.basic_reject(delivery_info.delivery_tag, false)
end

@example Requeue a message fetched via basic.get

conn  = Bunny.new
conn.start

ch    = conn.create_channel
# we assume the queue exists and has messages
delivery_info, properties, payload = ch.basic_get("bunny.examples.queue3", :manual_ack => true)
ch.basic_reject(delivery_info.delivery_tag, true)

@see Bunny::Channel#basic_nack @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 719
def basic_reject(delivery_tag, requeue = false)
  guarding_against_stale_delivery_tags(delivery_tag) do
    raise_if_no_longer_open!
    @connection.send_frame(AMQ::Protocol::Basic::Reject.encode(@id, delivery_tag, requeue))

    nil
  end
end
can_accept_queue_declare_ok?(method) click to toggle source

@private

# File lib/bunny/channel.rb, line 1647
def can_accept_queue_declare_ok?(method)
  @pending_queue_declare_name == method.queue ||
    pending_server_named_queue_declaration?
end
channel_flow(active) click to toggle source

Enables or disables message flow for the channel. When message flow is disabled, no new messages will be delivered to consumers on this channel. This is typically used by consumers that cannot keep up with the influx of messages.

@note Recent (e.g. 2.8.x., 3.x) RabbitMQ will employ TCP/IP-level back pressure on publishers if it detects

that consumers do not keep up with them.

@return [AMQ::Protocol::Channel::FlowOk] RabbitMQ response @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 1326
def channel_flow(active)
  raise_if_no_longer_open!

  @connection.send_frame(AMQ::Protocol::Channel::Flow.encode(@id, active))
  with_continuation_timeout do
    @last_channel_flow_ok = wait_on_continuations
  end
  raise_if_continuation_resulted_in_a_channel_error!

  @last_channel_flow_ok
end
channel_level_exception_after_operation_that_has_no_response?(method) click to toggle source

@private

# File lib/bunny/channel.rb, line 1746
def channel_level_exception_after_operation_that_has_no_response?(method)
  method.reply_code == 406 && method.reply_text =~ /unknown delivery tag/
end
client() click to toggle source

@return [Bunny::Session] Connection this channel was opened on

# File lib/bunny/channel.rb, line 282
def client
  @connection
end
close() click to toggle source

Closes the channel. Closed channels can no longer be used (this includes associated {Bunny::Queue}, {Bunny::Exchange} and {Bunny::Consumer} instances. @api public

# File lib/bunny/channel.rb, line 245
def close
  # see bunny#528
  raise_if_no_longer_open!

  @connection.close_channel(self)
  @status = :closed
  @work_pool.shutdown
  maybe_kill_consumer_work_pool!
end
closed?() click to toggle source

@return [Boolean] true if this channel is closed (manually or because of an exception), false otherwise @api public

# File lib/bunny/channel.rb, line 263
def closed?
  @status == :closed
end
confirm_select(callback = nil) click to toggle source

Enables publisher confirms for the channel. @return [AMQ::Protocol::Confirm::SelectOk] RabbitMQ response @see wait_for_confirms @see unconfirmed_set @see nacked_set @see rubybunny.info/articles/extensions.html RabbitMQ Extensions guide @api public

# File lib/bunny/channel.rb, line 1415
def confirm_select(callback = nil)
  raise_if_no_longer_open!

  if @next_publish_seq_no == 0
    @confirms_continuations = new_continuation
    @unconfirmed_set        = Set.new
    @nacked_set             = Set.new
    @next_publish_seq_no    = 1
    @only_acks_received = true
  end

  @confirms_callback = callback

  @connection.send_frame(AMQ::Protocol::Confirm::Select.encode(@id, false))
  with_continuation_timeout do
    @last_confirm_select_ok = wait_on_continuations
  end
  raise_if_continuation_resulted_in_a_channel_error!
  @last_confirm_select_ok
end
consume(queue, consumer_tag = generate_consumer_tag, no_ack = false, exclusive = false, arguments = nil, &block)
Alias for: basic_consume
consume_with(consumer)
Alias for: basic_consume_with
default_exchange() click to toggle source

Provides access to the default exchange @see rubybunny.info/articles/exchanges.html Exchanges and Publishing guide @api public

# File lib/bunny/channel.rb, line 375
def default_exchange
  Exchange.default(self)
end
deregister_exchange(exchange) click to toggle source

@private

# File lib/bunny/channel.rb, line 1945
def deregister_exchange(exchange)
  @exchange_mutex.synchronize { @exchanges.delete(exchange.name) }
end
deregister_queue(queue) click to toggle source

@private

# File lib/bunny/channel.rb, line 1925
def deregister_queue(queue)
  @queue_mutex.synchronize { @queues.delete(queue.name) }
end
deregister_queue_named(name) click to toggle source

@private

# File lib/bunny/channel.rb, line 1930
def deregister_queue_named(name)
  @queue_mutex.synchronize { @queues.delete(name) }
end
direct(name, opts = {}) click to toggle source

Declares a direct exchange or looks it up in the cache of previously declared exchanges.

@param [String] name Exchange name @param [Hash] opts Exchange parameters

@option opts [Boolean] :durable (false) Should the exchange be durable? @option opts [Boolean] :auto_delete (false) Should the exchange be automatically deleted when no longer in use? @option opts [Hash] :arguments ({}) Optional exchange arguments (used by RabbitMQ extensions)

@return [Bunny::Exchange] Exchange instance @see rubybunny.info/articles/exchanges.html Exchanges and Publishing guide @see rubybunny.info/articles/extensions.html RabbitMQ Extensions to AMQP 0.9.1 guide @api public

# File lib/bunny/channel.rb, line 332
def direct(name, opts = {})
  find_exchange(name) || Exchange.new(self, :direct, name, opts)
end
exchange(name, opts = {}) click to toggle source

Declares a headers exchange or looks it up in the cache of previously declared exchanges.

@param [String] name Exchange name @param [Hash] opts Exchange parameters

@option opts [String,Symbol] :type (:direct) Exchange type, e.g. :fanout or “x-consistent-hash” @option opts [Boolean] :durable (false) Should the exchange be durable? @option opts [Boolean] :auto_delete (false) Should the exchange be automatically deleted when no longer in use? @option opts [Hash] :arguments ({}) Optional exchange arguments

@return [Bunny::Exchange] Exchange instance @see rubybunny.info/articles/exchanges.html Exchanges and Publishing guide @see rubybunny.info/articles/extensions.html RabbitMQ Extensions to AMQP 0.9.1 guide

# File lib/bunny/channel.rb, line 393
def exchange(name, opts = {})
  Exchange.new(self, opts.fetch(:type, :direct), name, opts)
end
exchange_bind(source, destination, opts = {}) click to toggle source

Binds an exchange to another exchange using exchange.bind AMQP 0.9.1 extension that RabbitMQ provides.

@param [String] source Source exchange name @param [String] destination Destination exchange name @param [Hash] opts Options

@option opts [String] routing_key (nil) Routing key used for binding @option opts [Hash] arguments ({}) Optional arguments

@return [AMQ::Protocol::Exchange::BindOk] RabbitMQ response @see rubybunny.info/articles/exchanges.html Exchanges and Publishing guide @see rubybunny.info/articles/bindings.html Bindings guide @see rubybunny.info/articles/extensions.html RabbitMQ Extensions guide @api public

# File lib/bunny/channel.rb, line 1237
def exchange_bind(source, destination, opts = {})
  raise_if_no_longer_open!

  source_name = if source.respond_to?(:name)
                  source.name
                else
                  source
                end

  destination_name = if destination.respond_to?(:name)
                       destination.name
                     else
                       destination
                     end

  @connection.send_frame(AMQ::Protocol::Exchange::Bind.encode(@id,
      destination_name,
      source_name,
      opts[:routing_key],
      false,
      opts[:arguments]))
  with_continuation_timeout do
    @last_exchange_bind_ok = wait_on_continuations
  end

  raise_if_continuation_resulted_in_a_channel_error!
  @last_exchange_bind_ok
end
exchange_declare(name, type, opts = {}) click to toggle source

Declares a exchange using exchange.declare AMQP 0.9.1 method.

@param [String] name The name of the exchange. Note that LF and CR characters

will be stripped from the value.

@param [String,Symbol] type Exchange type, e.g. :fanout or :topic @param [Hash] opts Exchange properties

@option opts [Boolean] durable (false) Should information about this exchange be persisted to disk so that it

can survive broker restarts? Typically set to true for long-lived exchanges.

@option opts [Boolean] auto_delete (false) Should this exchange be deleted when it is no longer used? @option opts [Boolean] passive (false) If true, exchange will be checked for existence. If it does not

exist, {Bunny::NotFound} will be raised.

@return [AMQ::Protocol::Exchange::DeclareOk] RabbitMQ response @see rubybunny.info/articles/exchanges.html Exchanges and Publishing guide @api public

# File lib/bunny/channel.rb, line 1174
def exchange_declare(name, type, opts = {})
  raise_if_no_longer_open!

  # strip trailing new line and carriage returns
  # just like RabbitMQ does
  safe_name = name.gsub(/[\r\n]/, "")
  @connection.send_frame(AMQ::Protocol::Exchange::Declare.encode(@id,
      safe_name,
      type.to_s,
      opts.fetch(:passive, false),
      opts.fetch(:durable, false),
      opts.fetch(:auto_delete, false),
      opts.fetch(:internal, false),
      false, # nowait
      opts[:arguments]))
  with_continuation_timeout do
    @last_exchange_declare_ok = wait_on_continuations
  end

  raise_if_continuation_resulted_in_a_channel_error!
  @last_exchange_declare_ok
end
exchange_delete(name, opts = {}) click to toggle source

Deletes a exchange using exchange.delete AMQP 0.9.1 method

@param [String] name Exchange name @param [Hash] opts Options

@option opts [Boolean] if_unused (false) Should this exchange be deleted only if it is no longer used

@return [AMQ::Protocol::Exchange::DeleteOk] RabbitMQ response @see rubybunny.info/articles/exchanges.html Exchanges and Publishing guide @api public

# File lib/bunny/channel.rb, line 1207
def exchange_delete(name, opts = {})
  raise_if_no_longer_open!

  @connection.send_frame(AMQ::Protocol::Exchange::Delete.encode(@id,
      name,
      opts[:if_unused],
      false))
  with_continuation_timeout do
    @last_exchange_delete_ok = wait_on_continuations
  end

  raise_if_continuation_resulted_in_a_channel_error!
  @last_exchange_delete_ok
end
exchange_unbind(source, destination, opts = {}) click to toggle source

Unbinds an exchange from another exchange using exchange.unbind AMQP 0.9.1 extension that RabbitMQ provides.

@param [String] source Source exchange name @param [String] destination Destination exchange name @param [Hash] opts Options

@option opts [String] routing_key (nil) Routing key used for binding @option opts [Hash] arguments ({}) Optional arguments

@return [AMQ::Protocol::Exchange::UnbindOk] RabbitMQ response @see rubybunny.info/articles/exchanges.html Exchanges and Publishing guide @see rubybunny.info/articles/bindings.html Bindings guide @see rubybunny.info/articles/extensions.html RabbitMQ Extensions guide @api public

# File lib/bunny/channel.rb, line 1281
def exchange_unbind(source, destination, opts = {})
  raise_if_no_longer_open!

  source_name = if source.respond_to?(:name)
                  source.name
                else
                  source
                end

  destination_name = if destination.respond_to?(:name)
                       destination.name
                     else
                       destination
                     end

  @connection.send_frame(AMQ::Protocol::Exchange::Unbind.encode(@id,
      destination_name,
      source_name,
      opts[:routing_key],
      false,
      opts[:arguments]))
  with_continuation_timeout do
    @last_exchange_unbind_ok = wait_on_continuations
  end

  raise_if_continuation_resulted_in_a_channel_error!
  @last_exchange_unbind_ok
end
fanout(name, opts = {}) click to toggle source

Declares a fanout exchange or looks it up in the cache of previously declared exchanges.

@param [String] name Exchange name @param [Hash] opts Exchange parameters

@option opts [Boolean] :durable (false) Should the exchange be durable? @option opts [Boolean] :auto_delete (false) Should the exchange be automatically deleted when no longer in use? @option opts [Hash] :arguments ({}) Optional exchange arguments (used by RabbitMQ extensions)

@return [Bunny::Exchange] Exchange instance @see rubybunny.info/articles/exchanges.html Exchanges and Publishing guide @see rubybunny.info/articles/extensions.html RabbitMQ Extensions to AMQP 0.9.1 guide @api public

# File lib/bunny/channel.rb, line 314
def fanout(name, opts = {})
  find_exchange(name) || Exchange.new(self, :fanout, name, opts)
end
find_exchange(name) click to toggle source

@private

# File lib/bunny/channel.rb, line 1955
def find_exchange(name)
  @exchange_mutex.synchronize { @exchanges[name] }
end
find_queue(name) click to toggle source

@private

# File lib/bunny/channel.rb, line 1940
def find_queue(name)
  @queue_mutex.synchronize { @queues[name] }
end
flow(active) click to toggle source

Flow control. When set to false, RabbitMQ will stop delivering messages on this channel.

@param [Boolean] active Should messages to consumers on this channel be delivered? @api public

# File lib/bunny/channel.rb, line 444
def flow(active)
  channel_flow(active)
end
frame_size() click to toggle source

@private

# File lib/bunny/channel.rb, line 287
def frame_size
  @connection.frame_max
end
generate_consumer_tag(name = "bunny") click to toggle source

Unique string supposed to be used as a consumer tag.

@return [String] Unique string. @api plugin

# File lib/bunny/channel.rb, line 1466
def generate_consumer_tag(name = "bunny")
  "#{name}-#{Time.now.to_i * 1000}-#{Kernel.rand(999_999_999_999)}"
end
handle_ack_or_nack(delivery_tag_before_offset, multiple, nack) click to toggle source

@private

# File lib/bunny/channel.rb, line 1789
def handle_ack_or_nack(delivery_tag_before_offset, multiple, nack)
  delivery_tag          = delivery_tag_before_offset + @delivery_tag_offset
  confirmed_range_start = multiple ? @delivery_tag_offset + @unconfirmed_set.min : delivery_tag
  confirmed_range_end   = delivery_tag
  confirmed_range       = (confirmed_range_start..confirmed_range_end)

  @unconfirmed_set_mutex.synchronize do
    if nack
      @nacked_set.merge(@unconfirmed_set & confirmed_range)
    end

    @unconfirmed_set.subtract(confirmed_range)

    @only_acks_received = (@only_acks_received && !nack)

    @confirms_continuations.push(true) if @unconfirmed_set.empty?

    if @confirms_callback
      confirmed_range.each { |tag| @confirms_callback.call(tag, false, nack) }
    end
  end
end
handle_basic_get_empty(basic_get_empty) click to toggle source

@private

# File lib/bunny/channel.rb, line 1757
def handle_basic_get_empty(basic_get_empty)
  @basic_get_continuations.push([nil, nil, nil])
end
handle_basic_get_ok(basic_get_ok, properties, content) click to toggle source

@private

# File lib/bunny/channel.rb, line 1751
def handle_basic_get_ok(basic_get_ok, properties, content)
  basic_get_ok.delivery_tag = VersionedDeliveryTag.new(basic_get_ok.delivery_tag, @recoveries_counter.get)
  @basic_get_continuations.push([basic_get_ok, properties, content])
end
handle_basic_return(basic_return, properties, content) click to toggle source

@private

# File lib/bunny/channel.rb, line 1778
def handle_basic_return(basic_return, properties, content)
  x = find_exchange(basic_return.exchange)

  if x
    x.handle_return(ReturnInfo.new(basic_return), MessageProperties.new(properties), content)
  else
    @logger.warn "Exchange #{basic_return.exchange} is not in channel #{@id}'s cache! Dropping returned message!"
  end
end
handle_frameset(basic_deliver, properties, content) click to toggle source

@private

# File lib/bunny/channel.rb, line 1762
def handle_frameset(basic_deliver, properties, content)
  consumer = @consumers[basic_deliver.consumer_tag]
  if consumer
    @work_pool.submit do
      begin
        consumer.call(DeliveryInfo.new(basic_deliver, consumer, self), MessageProperties.new(properties), content)
      rescue StandardError => e
        @uncaught_exception_handler.call(e, consumer) if @uncaught_exception_handler
      end
    end
  else
    @logger.warn "No consumer for tag #{basic_deliver.consumer_tag} on channel #{@id}!"
  end
end
handle_method(method) click to toggle source

@private

# File lib/bunny/channel.rb, line 1653
def handle_method(method)
  @logger.debug { "Channel#handle_frame on channel #{@id}: #{method.inspect}" }
  case method
  when AMQ::Protocol::Queue::DeclareOk then
    # safeguard against late arrivals of responses and
    # so on, see ruby-amqp/bunny#558
    if can_accept_queue_declare_ok?(method)
      @continuations.push(method)
    else
      if !pending_server_named_queue_declaration?
        # this response is for an outdated/overwritten
        # queue.declare, drop it
        @logger.warn "Received a queue.declare-ok response for a mismatching queue (#{method.queue} instead of #{@pending_queue_declare_name}) on channel #{@id} possibly due to a timeout, ignoring it"
      end
    end
  when AMQ::Protocol::Queue::DeleteOk then
    @continuations.push(method)
  when AMQ::Protocol::Queue::PurgeOk then
    @continuations.push(method)
  when AMQ::Protocol::Queue::BindOk then
    @continuations.push(method)
  when AMQ::Protocol::Queue::UnbindOk then
    @continuations.push(method)
  when AMQ::Protocol::Exchange::BindOk then
    @continuations.push(method)
  when AMQ::Protocol::Exchange::UnbindOk then
    @continuations.push(method)
  when AMQ::Protocol::Exchange::DeclareOk then
    @continuations.push(method)
  when AMQ::Protocol::Exchange::DeleteOk then
    @continuations.push(method)
  when AMQ::Protocol::Basic::QosOk then
    @continuations.push(method)
  when AMQ::Protocol::Basic::RecoverOk then
    @continuations.push(method)
  when AMQ::Protocol::Channel::FlowOk then
    @continuations.push(method)
  when AMQ::Protocol::Basic::ConsumeOk then
    @continuations.push(method)
  when AMQ::Protocol::Basic::Cancel then
    if consumer = @consumers[method.consumer_tag]
      @work_pool.submit do
        begin
          if recovers_cancelled_consumers?
            consumer.handle_cancellation(method)
            @logger.info "Automatically recovering cancelled consumer #{consumer.consumer_tag} on queue #{consumer.queue_name}"

            consume_with(consumer)
          else
            @consumers.delete(method.consumer_tag)
            consumer.handle_cancellation(method)
          end
        rescue Exception => e
          @logger.error "Got exception when notifying consumer #{method.consumer_tag} about cancellation!"
          @uncaught_exception_handler.call(e, consumer) if @uncaught_exception_handler
        end
      end
    else
      @logger.warn "No consumer for tag #{method.consumer_tag} on channel #{@id}!"
    end
  when AMQ::Protocol::Basic::CancelOk then
    @continuations.push(method)
    unregister_consumer(method.consumer_tag)
  when AMQ::Protocol::Tx::SelectOk, AMQ::Protocol::Tx::CommitOk, AMQ::Protocol::Tx::RollbackOk then
    @continuations.push(method)
  when AMQ::Protocol::Tx::SelectOk then
    @continuations.push(method)
  when AMQ::Protocol::Confirm::SelectOk then
    @continuations.push(method)
  when AMQ::Protocol::Basic::Ack then
    handle_ack_or_nack(method.delivery_tag, method.multiple, false)
  when AMQ::Protocol::Basic::Nack then
    handle_ack_or_nack(method.delivery_tag, method.multiple, true)
  when AMQ::Protocol::Channel::Close then
    closed!
    @connection.send_frame(AMQ::Protocol::Channel::CloseOk.encode(@id))

    # basic.ack, basic.reject, basic.nack. MK.
    if channel_level_exception_after_operation_that_has_no_response?(method)
      @on_error.call(self, method) if @on_error
    else
      @last_channel_error = instantiate_channel_level_exception(method)
      @continuations.push(method)
    end

  when AMQ::Protocol::Channel::CloseOk then
    @continuations.push(method)
  else
    raise "Do not know how to handle #{method.inspect} in Bunny::Channel#handle_method"
  end
end
headers(name, opts = {}) click to toggle source

Declares a headers exchange or looks it up in the cache of previously declared exchanges.

@param [String] name Exchange name @param [Hash] opts Exchange parameters

@option opts [Boolean] :durable (false) Should the exchange be durable? @option opts [Boolean] :auto_delete (false) Should the exchange be automatically deleted when no longer in use? @option opts [Hash] :arguments ({}) Optional exchange arguments

@return [Bunny::Exchange] Exchange instance @see rubybunny.info/articles/exchanges.html Exchanges and Publishing guide @see rubybunny.info/articles/extensions.html RabbitMQ Extensions to AMQP 0.9.1 guide @api public

# File lib/bunny/channel.rb, line 368
def headers(name, opts = {})
  find_exchange(name) || Exchange.new(self, :headers, name, opts)
end
increment_recoveries_counter() click to toggle source

@private

# File lib/bunny/channel.rb, line 1582
def increment_recoveries_counter
  @recoveries_counter.increment
end
inspect() click to toggle source
# File lib/bunny/channel.rb, line 1604
def inspect
  to_s
end
maybe_kill_consumer_work_pool!() click to toggle source

@private

# File lib/bunny/channel.rb, line 1913
def maybe_kill_consumer_work_pool!
  if @work_pool && @work_pool.running?
    @work_pool.kill
  end
end
maybe_pause_consumer_work_pool!() click to toggle source

@private

# File lib/bunny/channel.rb, line 1908
def maybe_pause_consumer_work_pool!
  @work_pool.pause if @work_pool && @work_pool.running?
end
maybe_start_consumer_work_pool!() click to toggle source

Starts consumer work pool. Lazily called by basic_consume to avoid creating new threads that won't do any real work for channels that do not register consumers (e.g. only used for publishing). MK. @private

# File lib/bunny/channel.rb, line 1901
def maybe_start_consumer_work_pool!
  if @work_pool && !@work_pool.running?
    @work_pool.start
  end
end
nack(delivery_tag, multiple = false, requeue = false) click to toggle source

Rejects a message. A rejected message can be requeued or dropped by RabbitMQ. This method is similar to {Bunny::Channel#reject} but supports rejecting multiple messages at once, and is usually preferred.

@param [Integer] delivery_tag Delivery tag to reject @param [Boolean] multiple (false) Should all unacknowledged messages up to this be rejected as well? @param [Boolean] requeue (false) Should this message be requeued instead of dropping it? @see Bunny::Channel#ack @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 496
def nack(delivery_tag, multiple = false, requeue = false)
  basic_nack(delivery_tag.to_i, multiple, requeue)
end
number() click to toggle source

@return [Integer] Channel id

# File lib/bunny/channel.rb, line 272
def number
  self.id
end
on_error(&block) click to toggle source

Defines a handler for errors that are not responses to a particular operations (e.g. basic.ack, basic.reject, basic.nack).

@api public

# File lib/bunny/channel.rb, line 1481
def on_error(&block)
  @on_error = block
end
on_uncaught_exception(&block) click to toggle source

Defines a handler for uncaught exceptions in consumers (e.g. delivered message handlers).

@api public

# File lib/bunny/channel.rb, line 1489
def on_uncaught_exception(&block)
  @uncaught_exception_handler = block
end
open() click to toggle source

Opens the channel and resets its internal state @return [Bunny::Channel] Self @api public

# File lib/bunny/channel.rb, line 228
def open
  @threads_waiting_on_continuations           = Set.new
  @threads_waiting_on_confirms_continuations  = Set.new
  @threads_waiting_on_basic_get_continuations = Set.new

  @connection.open_channel(self)
  # clear last channel error
  @last_channel_error = nil

  @status = :open

  self
end
open?() click to toggle source

@return [Boolean] true if this channel is open, false otherwise @api public

# File lib/bunny/channel.rb, line 257
def open?
  @status == :open
end
pending_server_named_queue_declaration?() click to toggle source

@private

# File lib/bunny/channel.rb, line 1642
def pending_server_named_queue_declaration?
  @pending_queue_declare_name && @pending_queue_declare_name.empty?
end
prefetch(count, global = false)
Alias for: basic_qos
queue(name = AMQ::Protocol::EMPTY_STRING, opts = {}) click to toggle source

Declares a queue or looks it up in the per-channel cache.

@param [String] name Queue name. Pass an empty string to declare a server-named queue (make RabbitMQ generate a unique name). @param [Hash] opts Queue properties and other options

@option opts [Boolean] :durable (false) Should this queue be durable? @option opts [Boolean] :auto-delete (false) Should this queue be automatically deleted when the last consumer disconnects? @option opts [Boolean] :exclusive (false) Should this queue be exclusive (only can be used by this connection, removed when the connection is closed)? @option opts [Boolean] :arguments ({}) Additional optional arguments (typically used by RabbitMQ extensions and plugins)

@return [Bunny::Queue] Queue that was declared or looked up in the cache @see rubybunny.info/articles/queues.html Queues and Consumers guide @see rubybunny.info/articles/extensions.html RabbitMQ Extensions guide @api public

# File lib/bunny/channel.rb, line 416
def queue(name = AMQ::Protocol::EMPTY_STRING, opts = {})
  throw ArgumentError.new("queue name must not be nil") if name.nil?

  q = find_queue(name) || Bunny::Queue.new(self, name, opts)

  register_queue(q)
end
queue_bind(name, exchange, opts = {}) click to toggle source

Binds a queue to an exchange using queue.bind AMQP 0.9.1 method

@param [String] name Queue name @param [String] exchange Exchange name @param [Hash] opts Options

@option opts [String] routing_key (nil) Routing key used for binding @option opts [Hash] arguments ({}) Optional arguments

@return [AMQ::Protocol::Queue::BindOk] RabbitMQ response @see rubybunny.info/articles/queues.html Queues and Consumers guide @see rubybunny.info/articles/bindings.html Bindings guide @api public

# File lib/bunny/channel.rb, line 1095
def queue_bind(name, exchange, opts = {})
  raise_if_no_longer_open!

  exchange_name = if exchange.respond_to?(:name)
                    exchange.name
                  else
                    exchange
                  end

  @connection.send_frame(AMQ::Protocol::Queue::Bind.encode(@id,
      name,
      exchange_name,
      (opts[:routing_key] || opts[:key]),
      false,
      opts[:arguments]))
  with_continuation_timeout do
    @last_queue_bind_ok = wait_on_continuations
  end

  raise_if_continuation_resulted_in_a_channel_error!
  @last_queue_bind_ok
end
queue_declare(name, opts = {}) click to toggle source

Declares a queue using queue.declare AMQP 0.9.1 method.

@param [String] name The name of the queue or an empty string to let RabbitMQ generate a name.

Note that LF and CR characters will be stripped from the value.

@param [Hash] opts Queue properties

@option opts [Boolean] durable (false) Should information about this queue be persisted to disk so that it

can survive broker restarts? Typically set to true for long-lived queues.

@option opts [Boolean] auto_delete (false) Should this queue be deleted when the last consumer is cancelled? @option opts [Boolean] exclusive (false) Should only this connection be able to use this queue?

If true, the queue will be automatically deleted when this
connection is closed

@option opts [Boolean] passive (false) If true, queue will be checked for existence. If it does not

exist, {Bunny::NotFound} will be raised.

@return [AMQ::Protocol::Queue::DeclareOk] RabbitMQ response @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 1005
def queue_declare(name, opts = {})
  raise_if_no_longer_open!

  # strip trailing new line and carriage returns
  # just like RabbitMQ does
  safe_name = name.gsub(/[\r\n]/, "")
  @pending_queue_declare_name = safe_name
  @connection.send_frame(
    AMQ::Protocol::Queue::Declare.encode(@id,
      @pending_queue_declare_name,
      opts.fetch(:passive, false),
      opts.fetch(:durable, false),
      opts.fetch(:exclusive, false),
      opts.fetch(:auto_delete, false),
      false,
      opts[:arguments]))

  begin
    with_continuation_timeout do
      @last_queue_declare_ok = wait_on_continuations
    end
  ensure
    # clear pending continuation context if it belongs to us
    @pending_queue_declare_name = nil if @pending_queue_declare_name == safe_name
  end
  raise_if_continuation_resulted_in_a_channel_error!

  @last_queue_declare_ok
end
queue_delete(name, opts = {}) click to toggle source

Deletes a queue using queue.delete AMQP 0.9.1 method

@param [String] name Queue name @param [Hash] opts Options

@option opts [Boolean] if_unused (false) Should this queue be deleted only if it has no consumers? @option opts [Boolean] if_empty (false) Should this queue be deleted only if it has no messages?

@return [AMQ::Protocol::Queue::DeleteOk] RabbitMQ response @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 1046
def queue_delete(name, opts = {})
  raise_if_no_longer_open!

  @connection.send_frame(AMQ::Protocol::Queue::Delete.encode(@id,
      name,
      opts[:if_unused],
      opts[:if_empty],
      false))
  with_continuation_timeout do
    @last_queue_delete_ok = wait_on_continuations
  end
  raise_if_continuation_resulted_in_a_channel_error!

  @last_queue_delete_ok
end
queue_purge(name, opts = {}) click to toggle source

Purges a queue (removes all messages from it) using queue.purge AMQP 0.9.1 method.

@param [String] name Queue name

@return [AMQ::Protocol::Queue::PurgeOk] RabbitMQ response @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 1069
def queue_purge(name, opts = {})
  raise_if_no_longer_open!

  @connection.send_frame(AMQ::Protocol::Queue::Purge.encode(@id, name, false))

  with_continuation_timeout do
    @last_queue_purge_ok = wait_on_continuations
  end
  raise_if_continuation_resulted_in_a_channel_error!

  @last_queue_purge_ok
end
queue_unbind(name, exchange, opts = {}) click to toggle source

Unbinds a queue from an exchange using queue.unbind AMQP 0.9.1 method

@param [String] name Queue name @param [String] exchange Exchange name @param [Hash] opts Options

@option opts [String] routing_key (nil) Routing key used for binding @option opts [Hash] arguments ({}) Optional arguments

@return [AMQ::Protocol::Queue::UnbindOk] RabbitMQ response @see rubybunny.info/articles/queues.html Queues and Consumers guide @see rubybunny.info/articles/bindings.html Bindings guide @api public

# File lib/bunny/channel.rb, line 1131
def queue_unbind(name, exchange, opts = {})
  raise_if_no_longer_open!

  exchange_name = if exchange.respond_to?(:name)
                    exchange.name
                  else
                    exchange
                  end

  @connection.send_frame(AMQ::Protocol::Queue::Unbind.encode(@id,
      name,
      exchange_name,
      opts[:routing_key],
      opts[:arguments]))
  with_continuation_timeout do
    @last_queue_unbind_ok = wait_on_continuations
  end

  raise_if_continuation_resulted_in_a_channel_error!
  @last_queue_unbind_ok
end
read_and_reset_only_acks_received() click to toggle source

@private

# File lib/bunny/channel.rb, line 1872
def read_and_reset_only_acks_received
  @unconfirmed_set_mutex.synchronize do
    result = @only_acks_received
    @only_acks_received = true
    result
  end
end
read_next_frame(options = {}) click to toggle source

@private

# File lib/bunny/channel.rb, line 1920
def read_next_frame(options = {})
  @connection.read_next_frame(options = {})
end
recover(ignored = true) click to toggle source

Tells RabbitMQ to redeliver unacknowledged messages @api public

# File lib/bunny/channel.rb, line 450
def recover(ignored = true)
  # RabbitMQ only supports basic.recover with requeue = true
  basic_recover(true)
end
recover_cancelled_consumers!() click to toggle source

@api public

# File lib/bunny/channel.rb, line 1587
def recover_cancelled_consumers!
  @recover_cancelled_consumers = true
end
recover_confirm_mode() click to toggle source

Recovers publisher confirms mode. Used by the Automatic Network Failure Recovery feature.

@api plugin

# File lib/bunny/channel.rb, line 1529
def recover_confirm_mode
  if using_publisher_confirmations?
    @unconfirmed_set.clear
    @delivery_tag_offset = @next_publish_seq_no - 1
    confirm_select(@confirms_callback)
  end
end
recover_consumers() click to toggle source

Recovers consumers. Used by the Automatic Network Failure Recovery feature.

@api plugin

# File lib/bunny/channel.rb, line 1570
def recover_consumers
  unless @consumers.empty?
    @work_pool = ConsumerWorkPool.new(@work_pool.size, @work_pool.abort_on_exception)
    @work_pool.start
  end

  @consumer_mutex.synchronize { @consumers.values }.each do |c|
    c.recover_from_network_failure
  end
end
recover_exchanges() click to toggle source

Recovers exchanges. Used by the Automatic Network Failure Recovery feature.

@api plugin

# File lib/bunny/channel.rb, line 1549
def recover_exchanges
  @exchange_mutex.synchronize { @exchanges.values }.each do |x|
    x.recover_from_network_failure
  end
end
recover_from_network_failure() click to toggle source

Recovers basic.qos setting, exchanges, queues and consumers. Used by the Automatic Network Failure Recovery feature.

@api plugin

# File lib/bunny/channel.rb, line 1503
def recover_from_network_failure
  @logger.debug { "Recovering channel #{@id} after network failure" }
  release_all_continuations

  recover_prefetch_setting
  recover_confirm_mode
  recover_tx_mode
  recover_exchanges
  # this includes recovering bindings
  recover_queues
  recover_consumers
  increment_recoveries_counter
end
recover_prefetch_setting() click to toggle source

Recovers basic.qos setting. Used by the Automatic Network Failure Recovery feature.

@api plugin

# File lib/bunny/channel.rb, line 1521
def recover_prefetch_setting
  basic_qos(@prefetch_count, @prefetch_global) if @prefetch_count
end
recover_queues() click to toggle source

Recovers queues and bindings. Used by the Automatic Network Failure Recovery feature.

@api plugin

# File lib/bunny/channel.rb, line 1559
def recover_queues
  @queue_mutex.synchronize { @queues.values }.each do |q|
    @logger.debug { "Recovering queue #{q.name}" }
    q.recover_from_network_failure
  end
end
recover_tx_mode() click to toggle source

Recovers transaction mode. Used by the Automatic Network Failure Recovery feature.

@api plugin

# File lib/bunny/channel.rb, line 1541
def recover_tx_mode
  tx_select if @tx_mode
end
recovers_cancelled_consumers?() click to toggle source

@api public

# File lib/bunny/channel.rb, line 1592
def recovers_cancelled_consumers?
  !!@recover_cancelled_consumers
end
register_consumer(consumer_tag, consumer) click to toggle source

@private

# File lib/bunny/channel.rb, line 1619
def register_consumer(consumer_tag, consumer)
  @consumer_mutex.synchronize do
    @consumers[consumer_tag] = consumer
  end
end
register_exchange(exchange) click to toggle source

@private

# File lib/bunny/channel.rb, line 1950
def register_exchange(exchange)
  @exchange_mutex.synchronize { @exchanges[exchange.name] = exchange }
end
register_queue(queue) click to toggle source

@private

# File lib/bunny/channel.rb, line 1935
def register_queue(queue)
  @queue_mutex.synchronize { @queues[queue.name] = queue }
end
reject(delivery_tag, requeue = false) click to toggle source

Rejects a message. A rejected message can be requeued or dropped by RabbitMQ.

@param [Integer] delivery_tag Delivery tag to reject @param [Boolean] requeue Should this message be requeued instead of dropping it? @see Bunny::Channel#ack @see Bunny::Channel#nack @see rubybunny.info/articles/queues.html Queues and Consumers guide @api public

# File lib/bunny/channel.rb, line 470
def reject(delivery_tag, requeue = false)
  basic_reject(delivery_tag.to_i, requeue)
end
release_all_continuations() click to toggle source

Releases all continuations. Used by automatic network recovery. @private

# File lib/bunny/channel.rb, line 1883
def release_all_continuations
  @threads_waiting_on_confirms_continuations.each do |t|
    t.run
  end
  @threads_waiting_on_continuations.each do |t|
    t.run
  end
  @threads_waiting_on_basic_get_continuations.each do |t|
    t.run
  end

  self.reset_continuations
end
synchronize(&block) click to toggle source

Synchronizes given block using this channel's mutex. @api public

# File lib/bunny/channel.rb, line 1458
def synchronize(&block)
  @publishing_mutex.synchronize(&block)
end
temporary_queue(opts = {}) click to toggle source

Declares a new server-named queue that is automatically deleted when the connection is closed.

@return [Bunny::Queue] Queue that was declared @see queue @api public

# File lib/bunny/channel.rb, line 430
def temporary_queue(opts = {})
  queue("", opts.merge(:exclusive => true))
end
to_s() click to toggle source

@return [String] Brief human-readable representation of the channel

# File lib/bunny/channel.rb, line 1600
def to_s
  "#<#{self.class.name}:#{object_id} @id=#{self.number} @connection=#{@connection.to_s}> @open=#{open?}"
end
topic(name, opts = {}) click to toggle source

Declares a topic exchange or looks it up in the cache of previously declared exchanges.

@param [String] name Exchange name @param [Hash] opts Exchange parameters

@option opts [Boolean] :durable (false) Should the exchange be durable? @option opts [Boolean] :auto_delete (false) Should the exchange be automatically deleted when no longer in use? @option opts [Hash] :arguments ({}) Optional exchange arguments (used by RabbitMQ extensions)

@return [Bunny::Exchange] Exchange instance @see rubybunny.info/articles/exchanges.html Exchanges and Publishing guide @see rubybunny.info/articles/extensions.html RabbitMQ Extensions to AMQP 0.9.1 guide @api public

# File lib/bunny/channel.rb, line 350
def topic(name, opts = {})
  find_exchange(name) || Exchange.new(self, :topic, name, opts)
end
tx_commit() click to toggle source

Commits current transaction @return [AMQ::Protocol::Tx::CommitOk] RabbitMQ response @api public

# File lib/bunny/channel.rb, line 1363
def tx_commit
  raise_if_no_longer_open!

  @connection.send_frame(AMQ::Protocol::Tx::Commit.encode(@id))
  with_continuation_timeout do
    @last_tx_commit_ok = wait_on_continuations
  end
  raise_if_continuation_resulted_in_a_channel_error!

  @last_tx_commit_ok
end
tx_rollback() click to toggle source

Rolls back current transaction @return [AMQ::Protocol::Tx::RollbackOk] RabbitMQ response @api public

# File lib/bunny/channel.rb, line 1378
def tx_rollback
  raise_if_no_longer_open!

  @connection.send_frame(AMQ::Protocol::Tx::Rollback.encode(@id))
  with_continuation_timeout do
    @last_tx_rollback_ok = wait_on_continuations
  end
  raise_if_continuation_resulted_in_a_channel_error!

  @last_tx_rollback_ok
end
tx_select() click to toggle source

Puts the channel into transaction mode (starts a transaction) @return [AMQ::Protocol::Tx::SelectOk] RabbitMQ response @api public

# File lib/bunny/channel.rb, line 1347
def tx_select
  raise_if_no_longer_open!

  @connection.send_frame(AMQ::Protocol::Tx::Select.encode(@id))
  with_continuation_timeout do
    @last_tx_select_ok = wait_on_continuations
  end
  raise_if_continuation_resulted_in_a_channel_error!
  @tx_mode = true

  @last_tx_select_ok
end
unregister_consumer(consumer_tag) click to toggle source

@private

# File lib/bunny/channel.rb, line 1626
def unregister_consumer(consumer_tag)
  @consumer_mutex.synchronize do
    @consumers.delete(consumer_tag)
  end
end
using_publisher_confirmations?() click to toggle source

@return [Boolean] true if this channel has Publisher Confirms enabled, false otherwise @api public

# File lib/bunny/channel.rb, line 1403
def using_publisher_confirmations?
  @next_publish_seq_no > 0
end
Also aliased as: using_publisher_confirms?
using_publisher_confirms?()
using_tx?() click to toggle source

@return [Boolean] true if this channel has transactions enabled

# File lib/bunny/channel.rb, line 1391
def using_tx?
  !!@tx_mode
end
wait_for_confirms() click to toggle source

Blocks calling thread until confirms are received for all currently unacknowledged published messages. Returns immediately if there are no outstanding confirms.

@return [Boolean] true if all messages were acknowledged positively since the last time this method was called, false otherwise @see confirm_select @see unconfirmed_set @see nacked_set @see rubybunny.info/articles/extensions.html RabbitMQ Extensions guide @api public

# File lib/bunny/channel.rb, line 1446
def wait_for_confirms
  wait_on_confirms_continuations
  read_and_reset_only_acks_received
end
wait_on_basic_get_continuations() click to toggle source

@private

# File lib/bunny/channel.rb, line 1831
def wait_on_basic_get_continuations
  if @connection.threaded
    t = Thread.current
    @threads_waiting_on_basic_get_continuations << t

    begin
      @basic_get_continuations.poll(@connection.continuation_timeout)
    ensure
      @threads_waiting_on_basic_get_continuations.delete(t)
    end
  else
    connection.reader_loop.run_once until @basic_get_continuations.length > 0

    @basic_get_continuations.pop
  end
end
wait_on_confirms_continuations() click to toggle source

@private

# File lib/bunny/channel.rb, line 1849
def wait_on_confirms_continuations
  raise_if_no_longer_open!

  if @connection.threaded
    t = Thread.current
    @threads_waiting_on_confirms_continuations << t

    begin
      while @unconfirmed_set_mutex.synchronize { !@unconfirmed_set.empty? }
        @confirms_continuations.poll(@connection.continuation_timeout)
      end
    ensure
      @threads_waiting_on_confirms_continuations.delete(t)
    end
  else
    unless @unconfirmed_set.empty?
      connection.reader_loop.run_once until @confirms_continuations.length > 0
      @confirms_continuations.pop
    end
  end
end
wait_on_continuations() click to toggle source

@private

# File lib/bunny/channel.rb, line 1813
def wait_on_continuations
  if @connection.threaded
    t = Thread.current
    @threads_waiting_on_continuations << t

    begin
      @continuations.poll(@connection.continuation_timeout)
    ensure
      @threads_waiting_on_continuations.delete(t)
    end
  else
    connection.reader_loop.run_once until @continuations.length > 0

    @continuations.pop
  end
end
wait_on_continuations_timeout() click to toggle source

@private

# File lib/bunny/channel.rb, line 221
def wait_on_continuations_timeout
  @connection.transport_write_timeout
end
with_continuation_timeout(&block) click to toggle source

@private

# File lib/bunny/channel.rb, line 1614
def with_continuation_timeout(&block)
  Bunny::Timeout.timeout(wait_on_continuations_timeout, ClientTimeout, &block)
end

Protected Instance Methods

closed!() click to toggle source

@private

# File lib/bunny/channel.rb, line 1962
def closed!
  @status = :closed
  @work_pool.shutdown
  @connection.release_channel_id(@id)
end
guarding_against_stale_delivery_tags(tag, &block) click to toggle source

@private

# File lib/bunny/channel.rb, line 2039
def guarding_against_stale_delivery_tags(tag, &block)
  case tag
    # if a fixnum was passed, execute unconditionally. MK.
  when Integer then
    block.call
    # versioned delivery tags should be checked to avoid
    # sending out stale (invalid) tags after channel was reopened
    # during network failure recovery. MK.
  when VersionedDeliveryTag then
    if !tag.stale?(@recoveries_counter.get)
      block.call
    end
  end
end
instantiate_channel_level_exception(frame) click to toggle source

@private

# File lib/bunny/channel.rb, line 1969
def instantiate_channel_level_exception(frame)
  case frame
  when AMQ::Protocol::Channel::Close then
    klass = case frame.reply_code
            when 403 then
              AccessRefused
            when 404 then
              NotFound
            when 405 then
              ResourceLocked
            when 406 then
              PreconditionFailed
            else
              ChannelLevelException
            end

    klass.new(frame.reply_text, self, frame)
  end
end
new_continuation() click to toggle source

@private

# File lib/bunny/channel.rb, line 2028
def new_continuation
  Concurrent::LinkedContinuationQueue.new
end
raise_if_channel_close!(method) click to toggle source

@private

# File lib/bunny/channel.rb, line 2006
def raise_if_channel_close!(method)
  if method && method.is_a?(AMQ::Protocol::Channel::Close)
    # basic.ack, basic.reject, basic.nack. MK.
    if channel_level_exception_after_operation_that_has_no_response?(method)
      @on_error.call(self, method) if @on_error
    else
      @last_channel_error = instantiate_channel_level_exception(method)
      raise @last_channel_error
    end
  end
end
raise_if_continuation_resulted_in_a_channel_error!() click to toggle source

@private

# File lib/bunny/channel.rb, line 1990
def raise_if_continuation_resulted_in_a_channel_error!
  raise @last_channel_error if @last_channel_error
end
raise_if_no_longer_open!() click to toggle source

@private

# File lib/bunny/channel.rb, line 1995
def raise_if_no_longer_open!
  if closed?
    if @last_channel_error
      raise ChannelAlreadyClosed.new("cannot use a closed channel! Channel id: #{@id}, closed due to a server-reported channel error: #{@last_channel_error.message}", self)
    else
      raise ChannelAlreadyClosed.new("cannot use a closed channel! Channel id: #{@id}", self)
    end
  end
end
reset_continuations() click to toggle source

@private

# File lib/bunny/channel.rb, line 2019
def reset_continuations
  @continuations           = new_continuation
  @confirms_continuations  = new_continuation
  @basic_get_continuations = new_continuation
end