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
-
{Bunny::Channel#queue} is used to declare queues. The rest of the API is in {Bunny::Queue}.
### Exchange
Operations In Higher-level API
-
{Bunny::Channel#topic} declares a topic exchange. The rest of the API is in {Bunny::Exchange}.
-
{Bunny::Channel#direct} declares a direct exchange.
-
{Bunny::Channel#fanout} declares a fanout exchange.
-
{Bunny::Channel#headers} declares a headers exchange.
-
{Bunny::Channel#default_exchange}
-
{Bunny::Channel#exchange} is used to declare exchanges with type specified as a symbol or string.
## 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
@return [Bunny::Session] AMQP connection this channel was opened on
@return [Hash<String, Bunny::Consumer>] Consumer
instances declared on this channel
@return [Hash<String, Bunny::Exchange>] Exchange
instances declared on this channel
@return [Integer] Channel
id
@return [Set<Integer>] Set of nacked message indexes that have been nacked
@return [Integer] Next publisher confirmations sequence index
@return [Integer] active basic.qos prefetch value
@return [Integer] active basic.qos prefetch global mode
@return [Hash<String, Bunny::Queue>] Queue
instances declared on this channel
@return [Symbol] Channel
status (:opening, :open, :closed)
@return [Set<Integer>] Set of published message indexes that are currently unconfirmed
@return [Bunny::ConsumerWorkPool] Thread pool delivered messages are dispatched to.
Public Class Methods
@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
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
@return [Boolean] true if this channel is open
# File lib/bunny/channel.rb, line 277 def active open? end
@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
@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
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
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
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
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
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
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
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
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
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
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
@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
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
@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
@return [Bunny::Session] Connection this channel was opened on
# File lib/bunny/channel.rb, line 282 def client @connection end
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
@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
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
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
@private
# File lib/bunny/channel.rb, line 1945 def deregister_exchange(exchange) @exchange_mutex.synchronize { @exchanges.delete(exchange.name) } end
@private
# File lib/bunny/channel.rb, line 1925 def deregister_queue(queue) @queue_mutex.synchronize { @queues.delete(queue.name) } end
@private
# File lib/bunny/channel.rb, line 1930 def deregister_queue_named(name) @queue_mutex.synchronize { @queues.delete(name) } end
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
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
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
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
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
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
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
@private
# File lib/bunny/channel.rb, line 1955 def find_exchange(name) @exchange_mutex.synchronize { @exchanges[name] } end
@private
# File lib/bunny/channel.rb, line 1940 def find_queue(name) @queue_mutex.synchronize { @queues[name] } end
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
@private
# File lib/bunny/channel.rb, line 287 def frame_size @connection.frame_max end
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
@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
@private
# File lib/bunny/channel.rb, line 1757 def handle_basic_get_empty(basic_get_empty) @basic_get_continuations.push([nil, nil, nil]) end
@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
@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
@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
@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
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
@private
# File lib/bunny/channel.rb, line 1582 def increment_recoveries_counter @recoveries_counter.increment end
# File lib/bunny/channel.rb, line 1604 def inspect to_s end
@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
@private
# File lib/bunny/channel.rb, line 1908 def maybe_pause_consumer_work_pool! @work_pool.pause if @work_pool && @work_pool.running? end
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
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
@return [Integer] Channel
id
# File lib/bunny/channel.rb, line 272 def number self.id end
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
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
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
@return [Boolean] true if this channel is open, false otherwise @api public
# File lib/bunny/channel.rb, line 257 def open? @status == :open end
@private
# File lib/bunny/channel.rb, line 1642 def pending_server_named_queue_declaration? @pending_queue_declare_name && @pending_queue_declare_name.empty? end
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
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
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
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
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
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
@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
@private
# File lib/bunny/channel.rb, line 1920 def read_next_frame(options = {}) @connection.read_next_frame(options = {}) end
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
@api public
# File lib/bunny/channel.rb, line 1587 def recover_cancelled_consumers! @recover_cancelled_consumers = true end
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
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
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
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
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
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
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
@api public
# File lib/bunny/channel.rb, line 1592 def recovers_cancelled_consumers? !!@recover_cancelled_consumers end
@private
# File lib/bunny/channel.rb, line 1619 def register_consumer(consumer_tag, consumer) @consumer_mutex.synchronize do @consumers[consumer_tag] = consumer end end
@private
# File lib/bunny/channel.rb, line 1950 def register_exchange(exchange) @exchange_mutex.synchronize { @exchanges[exchange.name] = exchange } end
@private
# File lib/bunny/channel.rb, line 1935 def register_queue(queue) @queue_mutex.synchronize { @queues[queue.name] = queue } end
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
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
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
@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
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
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
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
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
@private
# File lib/bunny/channel.rb, line 1626 def unregister_consumer(consumer_tag) @consumer_mutex.synchronize do @consumers.delete(consumer_tag) end end
@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
@return [Boolean] true if this channel has transactions enabled
# File lib/bunny/channel.rb, line 1391 def using_tx? !!@tx_mode end
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
@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
@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
@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
@private
# File lib/bunny/channel.rb, line 221 def wait_on_continuations_timeout @connection.transport_write_timeout end
@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
@private
# File lib/bunny/channel.rb, line 1962 def closed! @status = :closed @work_pool.shutdown @connection.release_channel_id(@id) end
@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
@private
# File lib/bunny/channel.rb, line 2028 def new_continuation Concurrent::LinkedContinuationQueue.new end
@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
@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
@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
@private
# File lib/bunny/channel.rb, line 2019 def reset_continuations @continuations = new_continuation @confirms_continuations = new_continuation @basic_get_continuations = new_continuation end