Class: Bunny::Channel
- Inherits:
-
Object
- Object
- Bunny::Channel
- Defined in:
- lib/bunny/channel.rb
Overview
Channels in RabbitMQ
To quote 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 #close. Channels that get a channel-level exception are closed, too. Closed channels can no longer be used. Attempts to use them will raise ChannelAlreadyClosed.
ch = conn.create_channel
ch.close
Higher-level API
Bunny offers two sets of methods on Channel: known as higher-level and lower-level APIs, respectively. Higher-level API mimics amqp gem API where exchanges and queues are objects (instance of Exchange and 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, Langohr and Pika).
Queue Operations In Higher-level API
Exchange Operations In Higher-level API
- #topic declares a topic exchange. The rest of the API is in Exchange.
- #direct declares a direct exchange.
- #fanout declares a fanout exchange.
- #headers declares a headers exchange.
- #default_exchange
- #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 #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,
NotFound, that provide access to the underlying channel.close
method
information.
Consumer and Message operations (basic.*) collapse
- MAX_PREFETCH_COUNT =
prefetch_count is of type short in the protocol. MK.
(2 ** 16) - 1
Constant Summary collapse
- DEFAULT_CONTENT_TYPE =
"application/octet-stream".freeze
- SHORTSTR_LIMIT =
255
Instance Attribute Summary collapse
-
#connection ⇒ Bunny::Session
readonly
AMQP connection this channel was opened on.
-
#consumers ⇒ Hash<String, Bunny::Consumer>
readonly
Consumer instances declared on this channel.
-
#delivery_tag_offset ⇒ Integer
readonly
This will be set to the current sequence index during automatic network failure recovery to keep the sequence monotonic for the user and abstract the reset from the protocol.
-
#exchanges ⇒ Hash<String, Bunny::Exchange>
readonly
Exchange instances declared on this channel.
-
#id ⇒ Integer
Channel id.
-
#nacked_set ⇒ Set<Integer>
readonly
Set of nacked message indexes that have been nacked.
-
#next_publish_seq_no ⇒ Integer
readonly
Next publisher confirmations sequence index.
-
#prefetch_count ⇒ Integer
readonly
Active basic.qos prefetch value.
-
#prefetch_global ⇒ Integer
readonly
Active basic.qos prefetch global mode.
-
#queues ⇒ Hash<String, Bunny::Queue>
readonly
Queue instances declared on this channel.
-
#recoveries_counter ⇒ Object
readonly
Returns the value of attribute recoveries_counter.
-
#status ⇒ Symbol
readonly
Channel status (:opening, :open, :closed).
-
#unconfirmed_set ⇒ Set<Integer>
readonly
Set of published message indexes that are currently unconfirmed.
-
#work_pool ⇒ Bunny::ConsumerWorkPool
readonly
Thread pool delivered messages are dispatched to.
Backwards compatibility with 0.8.0 collapse
-
#active ⇒ Boolean
True if this channel is open.
-
#client ⇒ Bunny::Session
Connection this channel was opened on.
-
#number ⇒ Integer
Channel id.
Higher-level API for exchange operations collapse
-
#default_exchange ⇒ Object
Provides access to the default exchange.
-
#direct(name, opts = {}) ⇒ Bunny::Exchange
Declares a direct exchange or looks it up in the cache of previously declared exchanges.
-
#exchange(name, opts = {}) ⇒ Bunny::Exchange
Declares a headers exchange or looks it up in the cache of previously declared exchanges.
-
#fanout(name, opts = {}) ⇒ Bunny::Exchange
Declares a fanout exchange or looks it up in the cache of previously declared exchanges.
-
#headers(name, opts = {}) ⇒ Bunny::Exchange
Declares a headers exchange or looks it up in the cache of previously declared exchanges.
-
#topic(name, opts = {}) ⇒ Bunny::Exchange
Declares a topic exchange or looks it up in the cache of previously declared exchanges.
Higher-level API for queue operations collapse
-
#durable_queue(name, type = "classic", opts = {}) ⇒ Bunny::Queue
Declares a new server-named queue that is automatically deleted when the connection is closed.
-
#queue(name = AMQ::Protocol::EMPTY_STRING, opts = {}) ⇒ Bunny::Queue
Declares a queue or looks it up in the per-channel cache.
-
#quorum_queue(name, opts = {}) ⇒ Bunny::Queue
Declares a new client-named quorum queue.
-
#stream(name, opts = {}) ⇒ Bunny::Queue
Declares a new client-named stream (that Bunny can use as if it was a queue).
-
#temporary_queue(opts = {}) ⇒ Bunny::Queue
Declares a new server-named queue that is automatically deleted when the connection is closed.
QoS and Flow Control collapse
-
#flow(active) ⇒ Object
Flow control.
-
#recover(ignored = true) ⇒ Object
Tells RabbitMQ to redeliver unacknowledged messages.
Message acknowledgements collapse
-
#ack(delivery_tag, multiple = false) ⇒ Object
(also: #acknowledge)
Acknowledges a message.
-
#nack(delivery_tag, multiple = false, requeue = false) ⇒ Object
Rejects a message.
-
#reject(delivery_tag, requeue = false) ⇒ Object
Rejects a message.
Consumer and Message operations (basic.*) collapse
-
#any_consumers? ⇒ Boolean
True if there are consumers on this channel.
-
#basic_ack(delivery_tag, multiple = false) ⇒ NilClass
Acknowledges a delivery (message).
-
#basic_cancel(consumer_tag) ⇒ AMQ::Protocol::Basic::CancelOk
Removes a consumer.
-
#basic_consume(queue, consumer_tag = generate_consumer_tag, no_ack = false, exclusive = false, arguments = nil, &block) ⇒ AMQ::Protocol::Basic::ConsumeOk
(also: #consume)
Registers a consumer for queue.
-
#basic_consume_with(consumer) ⇒ AMQ::Protocol::Basic::ConsumeOk
(also: #consume_with)
Registers a consumer for queue as Consumer instance.
-
#basic_get(queue, opts = {:manual_ack => true}) ⇒ Array
Synchronously fetches a message from the queue, if there are any.
-
#basic_nack(delivery_tag, multiple = false, requeue = false) ⇒ NilClass
Rejects or requeues messages just like #basic_reject but can do so with multiple messages at once.
-
#basic_publish(payload, exchange, routing_key, opts = {}) ⇒ Bunny::Channel
Publishes a message using basic.publish AMQP 0.9.1 method.
-
#basic_qos(count, global = false) ⇒ AMQ::Protocol::Basic::QosOk
(also: #prefetch)
Controls message delivery rate using basic.qos AMQP 0.9.1 method.
-
#basic_recover(requeue) ⇒ AMQ::Protocol::Basic::RecoverOk
Redeliver unacknowledged messages.
-
#basic_reject(delivery_tag, requeue = false) ⇒ NilClass
Rejects or requeues a message.
Queue operations (queue.*) collapse
-
#queue_bind(name, exchange, opts = {}) ⇒ AMQ::Protocol::Queue::BindOk
Binds a queue to an exchange using queue.bind AMQP 0.9.1 method.
-
#queue_declare(name, opts = {}) ⇒ AMQ::Protocol::Queue::DeclareOk
Declares a queue using queue.declare AMQP 0.9.1 method.
-
#queue_delete(name, opts = {}) ⇒ AMQ::Protocol::Queue::DeleteOk
Deletes a queue using queue.delete AMQP 0.9.1 method.
-
#queue_purge(name, opts = {}) ⇒ AMQ::Protocol::Queue::PurgeOk
Purges a queue (removes all messages from it) using queue.purge AMQP 0.9.1 method.
-
#queue_unbind(name, exchange, opts = {}) ⇒ AMQ::Protocol::Queue::UnbindOk
Unbinds a queue from an exchange using queue.unbind AMQP 0.9.1 method.
Exchange operations (exchange.*) collapse
-
#exchange_bind(source, destination, opts = {}) ⇒ AMQ::Protocol::Exchange::BindOk
Binds an exchange to another exchange using exchange.bind AMQP 0.9.1 extension that RabbitMQ provides.
-
#exchange_declare(name, type, opts = {}) ⇒ AMQ::Protocol::Exchange::DeclareOk
Declares a exchange using exchange.declare AMQP 0.9.1 method.
-
#exchange_delete(name, opts = {}) ⇒ AMQ::Protocol::Exchange::DeleteOk
Deletes a exchange using exchange.delete AMQP 0.9.1 method.
-
#exchange_unbind(source, destination, opts = {}) ⇒ AMQ::Protocol::Exchange::UnbindOk
Unbinds an exchange from another exchange using exchange.unbind AMQP 0.9.1 extension that RabbitMQ provides.
Flow control (channel.*) collapse
-
#channel_flow(active) ⇒ AMQ::Protocol::Channel::FlowOk
Enables or disables message flow for the channel.
Transactions (tx.*) collapse
-
#tx_commit ⇒ AMQ::Protocol::Tx::CommitOk
Commits current transaction.
-
#tx_rollback ⇒ AMQ::Protocol::Tx::RollbackOk
Rolls back current transaction.
-
#tx_select ⇒ AMQ::Protocol::Tx::SelectOk
Puts the channel into transaction mode (starts a transaction).
-
#using_tx? ⇒ Boolean
True if this channel has transactions enabled.
Publisher Confirms (confirm.*) collapse
-
#confirm_select(callback = nil) ⇒ AMQ::Protocol::Confirm::SelectOk
Enables publisher confirms for the channel.
-
#using_publisher_confirmations? ⇒ Boolean
(also: #using_publisher_confirms?)
True if this channel has Publisher Confirms enabled, false otherwise.
-
#wait_for_confirms ⇒ Boolean
Blocks calling thread until confirms are received for all currently unacknowledged published messages.
Misc collapse
-
#generate_consumer_tag(name = "bunny") ⇒ String
Unique string supposed to be used as a consumer tag.
-
#synchronize(&block) ⇒ Object
Synchronizes given block using this channel’s mutex.
Network Failure Recovery collapse
-
#recover_cancelled_consumers! ⇒ Object
-
#recover_confirm_mode ⇒ Object
Recovers publisher confirms mode.
-
#recover_consumers ⇒ Object
Recovers consumers.
-
#recover_exchanges ⇒ Object
Recovers exchanges.
-
#recover_from_network_failure ⇒ Object
Recovers basic.qos setting, exchanges, queues and consumers.
-
#recover_prefetch_setting ⇒ Object
Recovers basic.qos setting.
-
#recover_queues ⇒ Object
Recovers queues and bindings.
-
#recover_tx_mode ⇒ Object
Recovers transaction mode.
-
#recovers_cancelled_consumers? ⇒ Boolean
Instance Method Summary collapse
-
#close ⇒ Object
Closes the channel.
-
#closed? ⇒ Boolean
True if this channel is closed (manually or because of an exception), false otherwise.
-
#initialize(connection = nil, id = nil, work_pool = ConsumerWorkPool.new(1)) ⇒ Channel
constructor
A new instance of Channel.
-
#inspect ⇒ Object
-
#on_error(&block) ⇒ Object
Defines a handler for errors that are not responses to a particular operations (e.g. basic.ack, basic.reject, basic.nack).
-
#on_uncaught_exception(&block) ⇒ Object
Defines a handler for uncaught exceptions in consumers (e.g. delivered message handlers).
-
#open ⇒ Bunny::Channel
Opens the channel and resets its internal state.
-
#open? ⇒ Boolean
True if this channel is open, false otherwise.
-
#to_s ⇒ String
Brief human-readable representation of the channel.
Constructor Details
#initialize(connection = nil, id = nil, work_pool = ConsumerWorkPool.new(1)) ⇒ Channel
Returns a new instance of Channel.
172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 |
# File 'lib/bunny/channel.rb', line 172 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 |
Instance Attribute Details
#connection ⇒ Bunny::Session (readonly)
Returns AMQP connection this channel was opened on.
139 140 141 |
# File 'lib/bunny/channel.rb', line 139 def connection @connection end |
#consumers ⇒ Hash<String, Bunny::Consumer> (readonly)
Returns Consumer instances declared on this channel.
159 160 161 |
# File 'lib/bunny/channel.rb', line 159 def consumers @consumers end |
#delivery_tag_offset ⇒ Integer (readonly)
This will be set to the current sequence index during automatic network failure recovery to keep the sequence monotonic for the user and abstract the reset from the protocol
149 150 151 |
# File 'lib/bunny/channel.rb', line 149 def delivery_tag_offset @delivery_tag_offset end |
#exchanges ⇒ Hash<String, Bunny::Exchange> (readonly)
Returns Exchange instances declared on this channel.
153 154 155 |
# File 'lib/bunny/channel.rb', line 153 def exchanges @exchanges end |
#id ⇒ Integer
Returns Channel id.
137 138 139 |
# File 'lib/bunny/channel.rb', line 137 def id @id end |
#nacked_set ⇒ Set<Integer> (readonly)
Returns Set of nacked message indexes that have been nacked.
157 158 159 |
# File 'lib/bunny/channel.rb', line 157 def nacked_set @nacked_set end |
#next_publish_seq_no ⇒ Integer (readonly)
Returns Next publisher confirmations sequence index.
145 146 147 |
# File 'lib/bunny/channel.rb', line 145 def next_publish_seq_no @next_publish_seq_no end |
#prefetch_count ⇒ Integer (readonly)
Returns active basic.qos prefetch value.
162 163 164 |
# File 'lib/bunny/channel.rb', line 162 def prefetch_count @prefetch_count end |
#prefetch_global ⇒ Integer (readonly)
Returns active basic.qos prefetch global mode.
164 165 166 |
# File 'lib/bunny/channel.rb', line 164 def prefetch_global @prefetch_global end |
#queues ⇒ Hash<String, Bunny::Queue> (readonly)
Returns Queue instances declared on this channel.
151 152 153 |
# File 'lib/bunny/channel.rb', line 151 def queues @queues end |
#recoveries_counter ⇒ Object (readonly)
Returns the value of attribute recoveries_counter.
223 224 225 |
# File 'lib/bunny/channel.rb', line 223 def recoveries_counter @recoveries_counter end |
#status ⇒ Symbol (readonly)
Returns Channel status (:opening, :open, :closed).
141 142 143 |
# File 'lib/bunny/channel.rb', line 141 def status @status end |
#unconfirmed_set ⇒ Set<Integer> (readonly)
Returns Set of published message indexes that are currently unconfirmed.
155 156 157 |
# File 'lib/bunny/channel.rb', line 155 def unconfirmed_set @unconfirmed_set end |
#work_pool ⇒ Bunny::ConsumerWorkPool (readonly)
Returns Thread pool delivered messages are dispatched to.
143 144 145 |
# File 'lib/bunny/channel.rb', line 143 def work_pool @work_pool end |
Instance Method Details
#ack(delivery_tag, multiple = false) ⇒ Object Also known as: acknowledge
Acknowledges a message. Acknowledged messages are completely removed from the queue.
553 554 555 |
# File 'lib/bunny/channel.rb', line 553 def ack(delivery_tag, multiple = false) basic_ack(delivery_tag.to_i, multiple) end |
#active ⇒ Boolean
Returns true if this channel is open.
282 283 284 |
# File 'lib/bunny/channel.rb', line 282 def active open? end |
#any_consumers? ⇒ Boolean
Returns true if there are consumers on this channel.
1050 1051 1052 |
# File 'lib/bunny/channel.rb', line 1050 def any_consumers? @consumer_mutex.synchronize { @consumers.any? } end |
#basic_ack(delivery_tag, multiple = false) ⇒ NilClass
Acknowledges a delivery (message).
839 840 841 842 843 844 845 846 |
# File 'lib/bunny/channel.rb', line 839 def basic_ack(delivery_tag, multiple = false) (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) ⇒ AMQ::Protocol::Basic::CancelOk
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.
1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 |
# File 'lib/bunny/channel.rb', line 1034 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) ⇒ AMQ::Protocol::Basic::ConsumeOk Also known as: consume
Registers a consumer for queue. Delivered messages will be handled with the block provided to this method.
926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 |
# File 'lib/bunny/channel.rb', line 926 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 |
#basic_consume_with(consumer) ⇒ AMQ::Protocol::Basic::ConsumeOk Also known as: consume_with
Registers a consumer for queue as Bunny::Consumer instance.
982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 |
# File 'lib/bunny/channel.rb', line 982 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 |
#basic_get(queue, opts = {:manual_ack => true}) ⇒ Array
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.
666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 |
# File 'lib/bunny/channel.rb', line 666 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) ⇒ NilClass
Rejects or requeues messages just like #basic_reject but can do so with multiple messages at once.
901 902 903 904 905 906 907 908 909 910 911 |
# File 'lib/bunny/channel.rb', line 901 def basic_nack(delivery_tag, multiple = false, requeue = false) (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 = {}) ⇒ Bunny::Channel
Publishes a message using basic.publish AMQP 0.9.1 method.
604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 |
# File 'lib/bunny/channel.rb', line 604 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) ⇒ AMQ::Protocol::Basic::QosOk Also known as: prefetch
Controls message delivery rate using basic.qos AMQP 0.9.1 method.
717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 |
# File 'lib/bunny/channel.rb', line 717 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 |
#basic_recover(requeue) ⇒ AMQ::Protocol::Basic::RecoverOk
Redeliver unacknowledged messages
741 742 743 744 745 746 747 748 749 750 751 |
# File 'lib/bunny/channel.rb', line 741 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) ⇒ NilClass
Rejects or requeues a message.
791 792 793 794 795 796 797 798 |
# File 'lib/bunny/channel.rb', line 791 def basic_reject(delivery_tag, requeue = false) (delivery_tag) do raise_if_no_longer_open! @connection.send_frame(AMQ::Protocol::Basic::Reject.encode(@id, delivery_tag, requeue)) nil end end |
#channel_flow(active) ⇒ AMQ::Protocol::Channel::FlowOk
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.
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.
1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 |
# File 'lib/bunny/channel.rb', line 1398 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 |
#client ⇒ Bunny::Session
Returns Connection this channel was opened on.
287 288 289 |
# File 'lib/bunny/channel.rb', line 287 def client @connection end |
#close ⇒ Object
Closes the channel. Closed channels can no longer be used (this includes associated Queue, Exchange and Bunny::Consumer instances.
250 251 252 253 254 255 256 257 258 |
# File 'lib/bunny/channel.rb', line 250 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? ⇒ Boolean
Returns true if this channel is closed (manually or because of an exception), false otherwise.
268 269 270 |
# File 'lib/bunny/channel.rb', line 268 def closed? @status == :closed end |
#confirm_select(callback = nil) ⇒ AMQ::Protocol::Confirm::SelectOk
Enables publisher confirms for the channel.
1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 |
# File 'lib/bunny/channel.rb', line 1487 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 |
#default_exchange ⇒ Object
Provides access to the default exchange
380 381 382 |
# File 'lib/bunny/channel.rb', line 380 def default_exchange @default_exchange ||= Exchange.default(self) end |
#direct(name, opts = {}) ⇒ Bunny::Exchange
Declares a direct exchange or looks it up in the cache of previously declared exchanges.
337 338 339 |
# File 'lib/bunny/channel.rb', line 337 def direct(name, opts = {}) find_exchange(name) || Exchange.new(self, :direct, name, opts) end |
#durable_queue(name, type = "classic", opts = {}) ⇒ Bunny::Queue
Declares a new server-named queue that is automatically deleted when the connection is closed.
480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 |
# File 'lib/bunny/channel.rb', line 480 def durable_queue(name, type = "classic", opts = {}) throw ArgumentError.new("queue name must not be nil") if name.nil? throw ArgumentError.new("queue name must not be empty (server-named durable queues do not make sense)") if name.empty? final_opts = opts.merge({ :type => type, :durable => true, # exclusive or auto-delete QQs do not make much sense :exclusive => false, :auto_delete => false }) q = find_queue(name) || Bunny::Queue.new(self, name, final_opts) register_queue(q) end |
#exchange(name, opts = {}) ⇒ Bunny::Exchange
Declares a headers exchange or looks it up in the cache of previously declared exchanges.
398 399 400 |
# File 'lib/bunny/channel.rb', line 398 def exchange(name, opts = {}) find_exchange(name) || Exchange.new(self, opts.fetch(:type, :direct), name, opts) end |
#exchange_bind(source, destination, opts = {}) ⇒ AMQ::Protocol::Exchange::BindOk
Binds an exchange to another exchange using exchange.bind AMQP 0.9.1 extension that RabbitMQ provides.
1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 |
# File 'lib/bunny/channel.rb', line 1309 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 = {}) ⇒ AMQ::Protocol::Exchange::DeclareOk
Declares a exchange using exchange.declare AMQP 0.9.1 method.
1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 |
# File 'lib/bunny/channel.rb', line 1246 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 = {}) ⇒ AMQ::Protocol::Exchange::DeleteOk
Deletes a exchange using exchange.delete AMQP 0.9.1 method
1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 |
# File 'lib/bunny/channel.rb', line 1279 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 = {}) ⇒ AMQ::Protocol::Exchange::UnbindOk
Unbinds an exchange from another exchange using exchange.unbind AMQP 0.9.1 extension that RabbitMQ provides.
1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 |
# File 'lib/bunny/channel.rb', line 1353 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 = {}) ⇒ Bunny::Exchange
Declares a fanout exchange or looks it up in the cache of previously declared exchanges.
319 320 321 |
# File 'lib/bunny/channel.rb', line 319 def fanout(name, opts = {}) find_exchange(name) || Exchange.new(self, :fanout, name, opts) end |
#flow(active) ⇒ Object
Flow control. When set to false, RabbitMQ will stop delivering messages on this channel.
516 517 518 |
# File 'lib/bunny/channel.rb', line 516 def flow(active) channel_flow(active) end |
#generate_consumer_tag(name = "bunny") ⇒ String
Unique string supposed to be used as a consumer tag.
1538 1539 1540 |
# File 'lib/bunny/channel.rb', line 1538 def generate_consumer_tag(name = "bunny") "#{name}-#{Time.now.to_i * 1000}-#{Kernel.rand(999_999_999_999)}" end |
#headers(name, opts = {}) ⇒ Bunny::Exchange
Declares a headers exchange or looks it up in the cache of previously declared exchanges.
373 374 375 |
# File 'lib/bunny/channel.rb', line 373 def headers(name, opts = {}) find_exchange(name) || Exchange.new(self, :headers, name, opts) end |
#inspect ⇒ Object
1679 1680 1681 |
# File 'lib/bunny/channel.rb', line 1679 def inspect to_s end |
#nack(delivery_tag, multiple = false, requeue = false) ⇒ Object
Rejects a message. A rejected message can be requeued or dropped by RabbitMQ. This method is similar to #reject but supports rejecting multiple messages at once, and is usually preferred.
568 569 570 |
# File 'lib/bunny/channel.rb', line 568 def nack(delivery_tag, multiple = false, requeue = false) basic_nack(delivery_tag.to_i, multiple, requeue) end |
#number ⇒ Integer
Returns Channel id.
277 278 279 |
# File 'lib/bunny/channel.rb', line 277 def number self.id end |
#on_error(&block) ⇒ Object
Defines a handler for errors that are not responses to a particular operations (e.g. basic.ack, basic.reject, basic.nack).
1553 1554 1555 |
# File 'lib/bunny/channel.rb', line 1553 def on_error(&block) @on_error = block end |
#on_uncaught_exception(&block) ⇒ Object
Defines a handler for uncaught exceptions in consumers (e.g. delivered message handlers).
1561 1562 1563 |
# File 'lib/bunny/channel.rb', line 1561 def on_uncaught_exception(&block) @uncaught_exception_handler = block end |
#open ⇒ Bunny::Channel
Opens the channel and resets its internal state
233 234 235 236 237 238 239 240 241 242 243 244 245 |
# File 'lib/bunny/channel.rb', line 233 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? ⇒ Boolean
Returns true if this channel is open, false otherwise.
262 263 264 |
# File 'lib/bunny/channel.rb', line 262 def open? @status == :open end |
#queue(name = AMQ::Protocol::EMPTY_STRING, opts = {}) ⇒ Bunny::Queue
Declares a queue or looks it up in the per-channel cache.
421 422 423 424 425 426 427 |
# File 'lib/bunny/channel.rb', line 421 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 = {}) ⇒ AMQ::Protocol::Queue::BindOk
Binds a queue to an exchange using queue.bind AMQP 0.9.1 method
1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 |
# File 'lib/bunny/channel.rb', line 1167 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 = {}) ⇒ AMQ::Protocol::Queue::DeclareOk
Declares a queue using queue.declare AMQP 0.9.1 method.
1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 |
# File 'lib/bunny/channel.rb', line 1077 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 = {}) ⇒ AMQ::Protocol::Queue::DeleteOk
Deletes a queue using queue.delete AMQP 0.9.1 method
1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 |
# File 'lib/bunny/channel.rb', line 1118 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 = {}) ⇒ AMQ::Protocol::Queue::PurgeOk
Purges a queue (removes all messages from it) using queue.purge AMQP 0.9.1 method.
1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 |
# File 'lib/bunny/channel.rb', line 1141 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 = {}) ⇒ AMQ::Protocol::Queue::UnbindOk
Unbinds a queue from an exchange using queue.unbind AMQP 0.9.1 method
1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 |
# File 'lib/bunny/channel.rb', line 1203 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 |
#quorum_queue(name, opts = {}) ⇒ Bunny::Queue
Declares a new client-named quorum queue.
440 441 442 443 444 445 |
# File 'lib/bunny/channel.rb', line 440 def quorum_queue(name, opts = {}) throw ArgumentError.new("quorum queue name must not be nil") if name.nil? throw ArgumentError.new("quorum queue name must not be empty (server-named QQs do not make sense)") if name.empty? durable_queue(name, Bunny::Queue::Types::QUORUM, opts) end |
#recover(ignored = true) ⇒ Object
Tells RabbitMQ to redeliver unacknowledged messages
522 523 524 525 |
# File 'lib/bunny/channel.rb', line 522 def recover(ignored = true) # RabbitMQ only supports basic.recover with requeue = true basic_recover(true) end |
#recover_cancelled_consumers! ⇒ Object
1662 1663 1664 |
# File 'lib/bunny/channel.rb', line 1662 def recover_cancelled_consumers! @recover_cancelled_consumers = true end |
#recover_confirm_mode ⇒ Object
Recovers publisher confirms mode. Used by the Automatic Network Failure Recovery feature. Set the offset to the previous publish sequence index as the protocol will reset the index to after recovery.
1602 1603 1604 1605 1606 1607 1608 1609 1610 |
# File 'lib/bunny/channel.rb', line 1602 def recover_confirm_mode if using_publisher_confirmations? @unconfirmed_set_mutex.synchronize do @unconfirmed_set.clear @delivery_tag_offset = @next_publish_seq_no - 1 end confirm_select(@confirms_callback) end end |
#recover_consumers ⇒ Object
Recovers consumers. Used by the Automatic Network Failure Recovery feature.
1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 |
# File 'lib/bunny/channel.rb', line 1645 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 ⇒ Object
Recovers exchanges. Used by the Automatic Network Failure Recovery feature.
1624 1625 1626 1627 1628 |
# File 'lib/bunny/channel.rb', line 1624 def recover_exchanges @exchange_mutex.synchronize { @exchanges.values }.each do |x| x.recover_from_network_failure end end |
#recover_from_network_failure ⇒ Object
Recovers basic.qos setting, exchanges, queues and consumers. Used by the Automatic Network Failure Recovery feature.
1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 |
# File 'lib/bunny/channel.rb', line 1575 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 ⇒ Object
Recovers basic.qos setting. Used by the Automatic Network Failure Recovery feature.
1593 1594 1595 |
# File 'lib/bunny/channel.rb', line 1593 def recover_prefetch_setting basic_qos(@prefetch_count, @prefetch_global) if @prefetch_count end |
#recover_queues ⇒ Object
Recovers queues and bindings. Used by the Automatic Network Failure Recovery feature.
1634 1635 1636 1637 1638 1639 |
# File 'lib/bunny/channel.rb', line 1634 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 ⇒ Object
Recovers transaction mode. Used by the Automatic Network Failure Recovery feature.
1616 1617 1618 |
# File 'lib/bunny/channel.rb', line 1616 def recover_tx_mode tx_select if @tx_mode end |
#recovers_cancelled_consumers? ⇒ Boolean
1667 1668 1669 |
# File 'lib/bunny/channel.rb', line 1667 def recovers_cancelled_consumers? !!@recover_cancelled_consumers end |
#reject(delivery_tag, requeue = false) ⇒ Object
Rejects a message. A rejected message can be requeued or dropped by RabbitMQ.
542 543 544 |
# File 'lib/bunny/channel.rb', line 542 def reject(delivery_tag, requeue = false) basic_reject(delivery_tag.to_i, requeue) end |
#stream(name, opts = {}) ⇒ Bunny::Queue
Declares a new client-named stream (that Bunny can use as if it was a queue). Note that Bunny would still use AMQP 0-9-1 to perform operations on this “queue”. To use stream-specific operations and to gain from stream protocol efficiency and partitioning, use a Ruby client for the RabbitMQ stream protocol.
462 463 464 465 466 467 |
# File 'lib/bunny/channel.rb', line 462 def stream(name, opts = {}) throw ArgumentError.new("stream name must not be nil") if name.nil? throw ArgumentError.new("stream name must not be empty (server-named QQs do not make sense)") if name.empty? durable_queue(name, Bunny::Queue::Types::STREAM, opts) end |
#synchronize(&block) ⇒ Object
Synchronizes given block using this channel’s mutex.
1530 1531 1532 |
# File 'lib/bunny/channel.rb', line 1530 def synchronize(&block) @publishing_mutex.synchronize(&block) end |
#temporary_queue(opts = {}) ⇒ Bunny::Queue
Declares a new server-named queue that is automatically deleted when the connection is closed.
502 503 504 |
# File 'lib/bunny/channel.rb', line 502 def temporary_queue(opts = {}) queue("", opts.merge(:exclusive => true)) end |
#to_s ⇒ String
Returns Brief human-readable representation of the channel.
1675 1676 1677 |
# File 'lib/bunny/channel.rb', line 1675 def to_s "#<#{self.class.name}:#{object_id} @id=#{self.number} @connection=#{@connection.to_s} @open=#{open?}>" end |
#topic(name, opts = {}) ⇒ Bunny::Exchange
Declares a topic exchange or looks it up in the cache of previously declared exchanges.
355 356 357 |
# File 'lib/bunny/channel.rb', line 355 def topic(name, opts = {}) find_exchange(name) || Exchange.new(self, :topic, name, opts) end |
#tx_commit ⇒ AMQ::Protocol::Tx::CommitOk
Commits current transaction
1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 |
# File 'lib/bunny/channel.rb', line 1435 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 ⇒ AMQ::Protocol::Tx::RollbackOk
Rolls back current transaction
1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 |
# File 'lib/bunny/channel.rb', line 1450 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 ⇒ AMQ::Protocol::Tx::SelectOk
Puts the channel into transaction mode (starts a transaction)
1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 |
# File 'lib/bunny/channel.rb', line 1419 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 |
#using_publisher_confirmations? ⇒ Boolean Also known as: using_publisher_confirms?
Returns true if this channel has Publisher Confirms enabled, false otherwise.
1475 1476 1477 |
# File 'lib/bunny/channel.rb', line 1475 def using_publisher_confirmations? @next_publish_seq_no > 0 end |
#using_tx? ⇒ Boolean
Returns true if this channel has transactions enabled.
1463 1464 1465 |
# File 'lib/bunny/channel.rb', line 1463 def using_tx? !!@tx_mode end |
#wait_for_confirms ⇒ Boolean
Blocks calling thread until confirms are received for all currently unacknowledged published messages. Returns immediately if there are no outstanding confirms.
1518 1519 1520 1521 |
# File 'lib/bunny/channel.rb', line 1518 def wait_for_confirms wait_on_confirms_continuations read_and_reset_only_acks_received end |