Class: Bunny::Session

Inherits:
Object
  • Object
show all
Defined in:
lib/bunny/session.rb

Overview

Represents AMQP 0.9.1 connection (to a RabbitMQ node).

See Also:

Constant Summary collapse

DEFAULT_HOST =

Default host used for connection

"127.0.0.1"
DEFAULT_VHOST =

Default virtual host used for connection

"/"
DEFAULT_USER =

Default username used for connection

"guest"
DEFAULT_PASSWORD =

Default password used for connection

"guest"
DEFAULT_HEARTBEAT =

Default heartbeat interval, the same value as RabbitMQ 3.0 uses.

:server
DEFAULT_FRAME_MAX = 
131072
CHANNEL_MAX_LIMIT =

Hard limit the user cannot go over regardless of server configuration.

65535
DEFAULT_CHANNEL_MAX = 
2047
CONNECT_TIMEOUT =

backwards compatibility

Transport::DEFAULT_CONNECTION_TIMEOUT
DEFAULT_CONTINUATION_TIMEOUT = 
15000
DEFAULT_CLIENT_PROPERTIES =

RabbitMQ client metadata

{
  capabilities: {
    publisher_confirms:           true,
    consumer_cancel_notify:       true,
    exchange_exchange_bindings:   true,
    :"basic.nack"                 => true,
    :"connection.blocked"         => true,
    # See http://www.rabbitmq.com/auth-notification.html
    authentication_failure_close: true
  },
  product:     "Bunny",
  platform:    ::RUBY_DESCRIPTION,
  version:     Bunny::VERSION,
  information: "https://github.com/ruby-amqp/bunny",
}
DEFAULT_LOCALE = 
"en_GB"
DEFAULT_NETWORK_RECOVERY_INTERVAL =

Default reconnection interval for TCP connection failures

5.0
DEFAULT_RECOVERABLE_EXCEPTIONS = 
[StandardError, TCPConnectionFailedForAllHosts, TCPConnectionFailed, AMQ::Protocol::EmptyResponseError, SystemCallError, Timeout::Error, Bunny::ConnectionLevelException, Bunny::ConnectionClosedError]

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection_string_or_opts = , optz = Hash.new) ⇒ Session

Returns a new instance of Session.

Parameters:

  • connection_string_or_opts (String, Hash) (defaults to: ) —

    Connection string or a hash of connection options

  • optz (Hash) (defaults to: Hash.new) —

    Extra options not related to connection

Options Hash (connection_string_or_opts):

  • :host (String) — default: "127.0.0.1" —

    Hostname or IP address to connect to

  • :hosts (Array<String>) — default: ["127.0.0.1"] —

    list of hostname or IP addresses to select hostname from when connecting

  • :addresses (Array<String>) — default: ["127.0.0.1:5672"] —

    list of addresses to select hostname and port from when connecting

  • :port (Integer) — default: 5672 —

    Port RabbitMQ listens on

  • :username (String) — default: "guest" —

    Username

  • :password (String) — default: "guest" —

    Password

  • :vhost (String) — default: "/" —

    Virtual host to use

  • :heartbeat (Integer, Symbol) — default: :server —

    Heartbeat timeout to offer to the server. :server means use the value suggested by RabbitMQ. 0 means heartbeats and socket read timeouts will be disabled (not recommended).

  • :network_recovery_interval (Integer) — default: 4 —

    Recovery interval periodic network recovery will use. This includes initial pause after network failure.

  • :tls (Boolean) — default: false —

    Should TLS/SSL be used?

  • :tls_cert (String) — default: nil —

    Path to client TLS/SSL certificate file (.pem)

  • :tls_key (String) — default: nil —

    Path to client TLS/SSL private key file (.pem)

  • :tls_ca_certificates (Array<String>) —

    Array of paths to TLS/SSL CA files (.pem), by default detected from OpenSSL configuration

  • :verify_peer (String) — default: true —

    Whether TLS peer verification should be performed

  • :tls_protocol (Symbol) — default: negotiated —

    What TLS version should be used (:TLSv1, :TLSv1_1, or :TLSv1_2)

  • :channel_max (Integer) — default: 2047 —

    Maximum number of channels allowed on this connection, minus 1 to account for the special channel 0.

  • :continuation_timeout (Integer) — default: 15000 —

    Timeout for client operations that expect a response (e.g. Queue#get), in milliseconds.

  • :connection_timeout (Integer) — default: 30 —

    Timeout in seconds for connecting to the server.

  • :read_timeout (Integer) — default: 30 —

    TCP socket read timeout in seconds. If heartbeats are disabled this will be ignored.

  • :write_timeout (Integer) — default: 30 —

    TCP socket write timeout in seconds.

  • :hosts_shuffle_strategy (Proc) —

    a callable that reorders a list of host strings, defaults to Array#shuffle

  • :recovery_completed (Proc) —

    a callable that will be called when a network recovery is performed

  • :logger (Logger) —

    The logger. If missing, one is created using :log_file and :log_level.

  • :log_file (IO, String) —

    The file or path to use when creating a logger. Defaults to STDOUT.

  • :logfile (IO, String) —

    DEPRECATED: use :log_file instead. The file or path to use when creating a logger. Defaults to STDOUT.

  • :log_level (Integer) —

    The log level to use when creating a logger. Defaults to LOGGER::WARN

  • :automatically_recover (Boolean) — default: true —

    Should automatically recover from network failures?

  • :recovery_attempts (Integer) — default: nil —

    Max number of recovery attempts, nil means forever

  • :reset_recovery_attempts_after_reconnection (Integer) — default: true —

    Should recovery attempt counter be reset after successful reconnection? When set to false, the attempt counter will last through the entire lifetime of the connection object.

  • :recovery_attempt_started (Proc) — default: nil —

    Will be called before every connection recovery attempt

  • :recovery_completed (Proc) — default: nil —

    Will be called after successful connection recovery

  • :recovery_attempts_exhausted (Proc) — default: nil —

    Will be called when the connection recovery failed after the specified amount of recovery attempts

  • :recover_from_connection_close (Boolean) — default: true —

    Should this connection recover after receiving a server-sent connection.close (e.g. connection was force closed)?

  • :session_error_handler (Object) — default: ExceptionAccumulator.new —

    Object which responds to #raise that will act as a session error handler. Defaults to a Bunny::ExceptionAccumulator instance which stores exceptions for safe retrieval later. Can be set to Thread.current for legacy behavior (raises asynchronous exceptions in the creating thread), or any object that responds to #raise.

  • :topology_recovery_filter (Bunny::TopologyRecoveryFilter) —

    if provided, will be used for object filtering during topology recovery

Options Hash (optz):

  • :auth_mechanism (String) — default: "PLAIN" —

    Authentication mechanism, PLAIN or EXTERNAL

  • :locale (String) — default: "PLAIN" —

    Locale RabbitMQ should use

  • :connection_name (String) — default: nil —

    Client-provided connection name, if any. Note that the value returned does not uniquely identify a connection and cannot be used as a connection identifier in HTTP API requests.

See Also:



144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
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
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
 
# File 'lib/bunny/session.rb', line 144

def initialize(connection_string_or_opts = ENV['RABBITMQ_URL'], optz = Hash.new)
  opts = case (connection_string_or_opts)
         when nil then
           Hash.new
         when String then
           self.class.parse_uri(connection_string_or_opts)
         when Hash then
           connection_string_or_opts
         end.merge(optz)

  @default_hosts_shuffle_strategy = Proc.new { |hosts| hosts.shuffle }

  @opts            = opts
  log_file         = opts[:log_file] || opts[:logfile] || STDOUT
  log_level        = opts[:log_level] || ENV["BUNNY_LOG_LEVEL"] || Logger::WARN
  # we might need to log a warning about ill-formatted IPv6 address but
  # progname includes hostname, so init like this first
  @logger          = opts.fetch(:logger, init_default_logger_without_progname(log_file, log_level))

  @addresses       = self.addresses_from(opts)
  @address_index   = 0

  @transport       = nil
  @user            = self.username_from(opts)
  @pass            = self.password_from(opts)
  @vhost           = self.vhost_from(opts)
  @threaded        = opts.fetch(:threaded, true)

  # re-init, see above
  @logger          = opts.fetch(:logger, init_default_logger(log_file, log_level))

  validate_connection_options(opts)
  @last_connection_error = nil

  # should automatic recovery from network failures be used?
  @automatically_recover = if opts[:automatically_recover].nil? && opts[:automatic_recovery].nil?
                             true
                           else
                             opts[:automatically_recover] | opts[:automatic_recovery]
                           end
  @recovering_from_network_failure = false
  @max_recovery_attempts = opts[:recovery_attempts]
  @recovery_attempts     = @max_recovery_attempts
  # When this is set, connection attempts won't be reset after
  # successful reconnection. Some find this behavior more sensible
  # than the per-failure attempt counter. MK.
  @reset_recovery_attempt_counter_after_reconnection = opts.fetch(:reset_recovery_attempts_after_reconnection, true)

  @network_recovery_interval = opts.fetch(:network_recovery_interval, DEFAULT_NETWORK_RECOVERY_INTERVAL)
  @recover_from_connection_close = opts.fetch(:recover_from_connection_close, true)
  # in ms
  @continuation_timeout   = opts.fetch(:continuation_timeout, DEFAULT_CONTINUATION_TIMEOUT)

  @status             = :not_connected
  @manually_closed    = false
  @blocked            = false

  # these are negotiated with the broker during the connection tuning phase
  @client_frame_max   = opts.fetch(:frame_max, DEFAULT_FRAME_MAX)
  @client_channel_max = normalize_client_channel_max(opts.fetch(:channel_max, DEFAULT_CHANNEL_MAX))
  # will be-renegotiated during connection tuning steps. MK.
  @channel_max        = @client_channel_max
  @heartbeat_sender   = nil
  @client_heartbeat   = self.heartbeat_from(opts)

  client_props         = opts[:properties] || opts[:client_properties] || {}
  @connection_name     = client_props[:connection_name] || opts[:connection_name]
  @client_properties   = DEFAULT_CLIENT_PROPERTIES.merge(client_props)
                                                  .merge(connection_name: connection_name)
  @mechanism           = normalize_auth_mechanism(opts.fetch(:auth_mechanism, "PLAIN"))
  @credentials_encoder = credentials_encoder_for(@mechanism)
  @locale              = @opts.fetch(:locale, DEFAULT_LOCALE)

  @mutex_impl          = @opts.fetch(:mutex_impl, Monitor)

  # mutex for the channel id => channel hash
  @channel_mutex       = @mutex_impl.new
  # transport operations/continuations mutex. A workaround for
  # the non-reentrant Ruby mutexes. MK.
  @transport_mutex     = @mutex_impl.new
  @status_mutex        = @mutex_impl.new
  @address_index_mutex = @mutex_impl.new
  @recovery_mutex      = @mutex_impl.new

  @channels            = Hash.new

  trf = @opts.fetch(:topology_recovery_filter, DefaultTopologyRecoveryFilter.new)
  @topology_registry = TopologyRegistry.new(topology_recovery_filter: trf)

  @recovery_attempt_started = opts[:recovery_attempt_started]
  @recovery_completed       = opts[:recovery_completed]
  @recovery_attempts_exhausted = opts[:recovery_attempts_exhausted]

  @session_error_handler = opts.fetch(:session_error_handler, ExceptionAccumulator.new)

  @recoverable_exceptions = DEFAULT_RECOVERABLE_EXCEPTIONS.dup

  self.reset_continuations
  self.initialize_transport

end

Instance Attribute Details

#channel_id_allocator ⇒ Object (readonly)

Returns the value of attribute channel_id_allocator.



83
84
85
 
# File 'lib/bunny/session.rb', line 83

def channel_id_allocator
  @channel_id_allocator
end

#channel_max ⇒ Object (readonly)

Returns the value of attribute channel_max.



81
82
83
 
# File 'lib/bunny/session.rb', line 81

def channel_max
  @channel_max
end

#connection_name ⇒ Object (readonly)

Returns the value of attribute connection_name.



94
95
96
 
# File 'lib/bunny/session.rb', line 94

def connection_name
  @connection_name
end

#continuation_timeout ⇒ Integer (readonly)

Returns Timeout for blocking protocol operations (queue.declare, queue.bind, etc), in milliseconds. Default is 15000.

Returns:

  • (Integer) —

    Timeout for blocking protocol operations (queue.declare, queue.bind, etc), in milliseconds. Default is 15000.



92
93
94
 
# File 'lib/bunny/session.rb', line 92

def continuation_timeout
  @continuation_timeout
end

#frame_max ⇒ Object (readonly)

Returns the value of attribute frame_max.



81
82
83
 
# File 'lib/bunny/session.rb', line 81

def frame_max
  @frame_max
end

#heartbeat ⇒ Object (readonly)

Returns the value of attribute heartbeat.



81
82
83
 
# File 'lib/bunny/session.rb', line 81

def heartbeat
  @heartbeat
end

#logger ⇒ Logger (readonly)

Returns:

  • (Logger) 


90
91
92
 
# File 'lib/bunny/session.rb', line 90

def logger
  @logger
end

#mechanism ⇒ String (readonly)

Authentication mechanism, e.g. "PLAIN" or "EXTERNAL"

Returns:

  • (String) 


88
89
90
 
# File 'lib/bunny/session.rb', line 88

def mechanism
  @mechanism
end

#mutex_impl ⇒ Object (readonly) 



302
303
304
 
# File 'lib/bunny/session.rb', line 302

def mutex_impl
  @mutex_impl
end

#network_recovery_interval ⇒ Object (readonly)

Returns the value of attribute network_recovery_interval.



93
94
95
 
# File 'lib/bunny/session.rb', line 93

def network_recovery_interval
  @network_recovery_interval
end

#pass ⇒ Object (readonly)

Returns the value of attribute pass.



81
82
83
 
# File 'lib/bunny/session.rb', line 81

def pass
  @pass
end

#recoverable_exceptions ⇒ Object

Returns the value of attribute recoverable_exceptions.



96
97
98
 
# File 'lib/bunny/session.rb', line 96

def recoverable_exceptions
  @recoverable_exceptions
end

#server_authentication_mechanisms ⇒ Object (readonly)

Returns the value of attribute server_authentication_mechanisms.



82
83
84
 
# File 'lib/bunny/session.rb', line 82

def server_authentication_mechanisms
  @server_authentication_mechanisms
end

#server_capabilities ⇒ Object (readonly)

Returns the value of attribute server_capabilities.



82
83
84
 
# File 'lib/bunny/session.rb', line 82

def server_capabilities
  @server_capabilities
end

#server_locales ⇒ Object (readonly)

Returns the value of attribute server_locales.



82
83
84
 
# File 'lib/bunny/session.rb', line 82

def server_locales
  @server_locales
end

#server_properties ⇒ Object (readonly)

Returns the value of attribute server_properties.



82
83
84
 
# File 'lib/bunny/session.rb', line 82

def server_properties
  @server_properties
end

#session_error_handler ⇒ Object (readonly)

Returns the session error handler. By default, this is a Bunny::ExceptionAccumulator instance.

Returns:

  • (Object) —

    the session error handler



519
520
521
 
# File 'lib/bunny/session.rb', line 519

def session_error_handler
  @session_error_handler
end

#socket_configurator ⇒ Object

Returns the value of attribute socket_configurator.



95
96
97
 
# File 'lib/bunny/session.rb', line 95

def socket_configurator
  @socket_configurator
end

#status ⇒ Object (readonly)

Returns the value of attribute status.



81
82
83
 
# File 'lib/bunny/session.rb', line 81

def status
  @status
end

#threaded ⇒ Object (readonly)

Returns the value of attribute threaded.



81
82
83
 
# File 'lib/bunny/session.rb', line 81

def threaded
  @threaded
end

#topology_registry ⇒ Bunny::TopologyRegistry (readonly)

Returns:



85
86
87
 
# File 'lib/bunny/session.rb', line 85

def topology_registry
  @topology_registry
end

#transport ⇒ Bunny::Transport (readonly)

Returns:



80
81
82
 
# File 'lib/bunny/session.rb', line 80

def transport
  @transport
end

#user ⇒ Object (readonly)

Returns the value of attribute user.



81
82
83
 
# File 'lib/bunny/session.rb', line 81

def user
  @user
end

#vhost ⇒ Object (readonly)

Returns the value of attribute vhost.



81
82
83
 
# File 'lib/bunny/session.rb', line 81

def vhost
  @vhost
end

Class Method Details

.parse_uri(uri) ⇒ Hash

Parses an amqp URI into a hash that #initialize accepts.

Parameters:

  • uri (String | Hash) —

    amqp or amqps URI to parse

Returns:

  • (Hash) —

    Parsed URI as a hash



568
569
570
 
# File 'lib/bunny/session.rb', line 568

def self.parse_uri(uri)
  AMQ::Settings.configure(uri)
end

Instance Method Details

#addresses_from(options) ⇒ Object 



1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
 
# File 'lib/bunny/session.rb', line 1316

def addresses_from(options)
  shuffle_strategy = options.fetch(:hosts_shuffle_strategy, @default_hosts_shuffle_strategy)

  addresses = options[:host] || options[:hostname] || options[:addresses] ||
    options[:hosts] || ["#{DEFAULT_HOST}:#{port_from(options)}"]
  addresses = [addresses] unless addresses.is_a? Array

  addrs = addresses.map do |address|
    host_with_port?(address) ? address : "#{address}:#{port_from(@opts)}"
  end

  shuffle_strategy.call(addrs)
end

#after_recovery_attempts_exhausted(&block) ⇒ Object

Defines a callable (e.g. a block) that will be called when the connection recovery failed after the specified numbers of recovery attempts.



629
630
631
 
# File 'lib/bunny/session.rb', line 629

def after_recovery_attempts_exhausted(&block)
  @recovery_attempts_exhausted = block
end

#after_recovery_completed(&block) ⇒ Object

Defines a callable (e.g. a block) that will be called after successful connection recovery.



622
623
624
 
# File 'lib/bunny/session.rb', line 622

def after_recovery_completed(&block)
  @recovery_completed = block
end

#announce_network_failure_recovery ⇒ Object 



996
997
998
999
1000
1001
1002
 
# File 'lib/bunny/session.rb', line 996

def announce_network_failure_recovery
  if recovery_attempts_limited?
    @logger.warn "Will recover from a network failure (#{@recovery_attempts} out of #{@max_recovery_attempts} left)..."
  else
    @logger.warn "Will recover from a network failure (no retry limit)..."
  end
end

#automatically_recover? ⇒ Boolean

Returns true if this connection has automatic recovery from network failure enabled.

Returns:

  • (Boolean) —

    true if this connection has automatic recovery from network failure enabled



482
483
484
 
# File 'lib/bunny/session.rb', line 482

def automatically_recover?
  @automatically_recover
end

#before_recovery_attempt_starts(&block) ⇒ Object

Defines a callable (e.g. a block) that will be called before every connection recovery attempt.



616
617
618
 
# File 'lib/bunny/session.rb', line 616

def before_recovery_attempt_starts(&block)
  @recovery_attempt_started = block
end

#blocked? ⇒ Boolean

Returns true if the connection is currently blocked by RabbitMQ because it's running low on RAM, disk space, or other resource; false otherwise.

Returns:

  • (Boolean) —

    true if the connection is currently blocked by RabbitMQ because it's running low on RAM, disk space, or other resource; false otherwise

See Also:



510
511
512
 
# File 'lib/bunny/session.rb', line 510

def blocked?
  @blocked
end

#clean_up_and_fail_on_connection_close!(method) ⇒ Object 



1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
 
# File 'lib/bunny/session.rb', line 1289

def clean_up_and_fail_on_connection_close!(method)
  @last_connection_error = instantiate_connection_level_exception(method)
  @continuations.push(method)

  clean_up_on_shutdown
  if threaded?
    @session_error_handler.raise(@last_connection_error)
  else
    raise @last_connection_error
  end
end

#clean_up_on_shutdown ⇒ Object 



1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
 
# File 'lib/bunny/session.rb', line 1301

def clean_up_on_shutdown
  begin
    shut_down_all_consumer_work_pools!
    maybe_shutdown_reader_loop
    maybe_shutdown_heartbeat_sender
  rescue ShutdownSignal => _sse
    # no-op
  rescue Exception => e
    @logger.warn "Caught an exception when cleaning up after receiving connection.close: #{e.message}"
  ensure
    close_transport
  end
end

#clear_exceptions ⇒ Array<Exception>

Clears all accumulated exceptions from the session error handler. Only works when using the default ExceptionAccumulator handler.

Returns:

  • (Array<Exception>) —

    the exceptions that were cleared

Raises:

  • (NoMethodError) —

    if the session error handler doesn't respond to #clear



547
548
549
 
# File 'lib/bunny/session.rb', line 547

def clear_exceptions
  @session_error_handler.clear
end

#close(await_response = true) ⇒ Object Also known as: stop

Closes the connection. This involves closing all of its channels.



414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
 
# File 'lib/bunny/session.rb', line 414

def close(await_response = true)
  @status_mutex.synchronize { @status = :closing }

  ignoring_io_errors do
    if @transport.open?
      @logger.debug "Transport is still open..."
      close_all_channels

      @logger.debug "Will close all channels...."
      self.close_connection(await_response)
    end

    clean_up_on_shutdown
  end
  @status_mutex.synchronize do
    @status = :closed
    @manually_closed = true
  end
  @logger.debug "Connection is closed"
  true
end

#close_all_channels ⇒ Object 



690
691
692
693
694
695
696
 
# File 'lib/bunny/session.rb', line 690

def close_all_channels
  @channel_mutex.synchronize do
    @channels.reject {|n, ch| n == 0 || !ch.open? }.each do |_, ch|
      Bunny::Timeout.timeout(@transport.disconnect_timeout, ClientTimeout) { ch.close }
    end
  end
end

#close_channel(ch) ⇒ Object 



665
666
667
668
669
670
671
672
673
674
675
676
677
 
# File 'lib/bunny/session.rb', line 665

def close_channel(ch)
  @channel_mutex.synchronize do
    n = ch.number

    @transport.send_frame(AMQ::Protocol::Channel::Close.encode(n, 200, "Goodbye", 0, 0))
    @last_channel_close_ok = wait_on_continuations
    raise_if_continuation_resulted_in_a_connection_error!

    self.unregister_channel(ch)
    self.release_channel_id(ch.id)
    @last_channel_close_ok
  end
end

#close_connection(await_response = true) ⇒ Object 



699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
 
# File 'lib/bunny/session.rb', line 699

def close_connection(await_response = true)
  if @transport.open?
    @logger.debug "Transport is still open"
    @transport.send_frame(AMQ::Protocol::Connection::Close.encode(200, "Goodbye", 0, 0))

    if await_response
      @logger.debug "Waiting for a connection.close-ok..."
      @last_connection_close_ok = wait_on_continuations
    end
  end

  shut_down_all_consumer_work_pools!
  maybe_shutdown_heartbeat_sender
  @status_mutex.synchronize { @status = :not_connected }
end

#close_transport ⇒ Object 



1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
 
# File 'lib/bunny/session.rb', line 1473

def close_transport
  begin
    @transport.close
  rescue StandardError => e
    @logger.error "Exception when closing transport:"
    @logger.error e.class.name
    @logger.error e.message
    @logger.error e.backtrace
  end
end

#closed? ⇒ Boolean

Returns true if this AMQP 0.9.1 connection is closed.

Returns:

  • (Boolean) —

    true if this AMQP 0.9.1 connection is closed



464
465
466
 
# File 'lib/bunny/session.rb', line 464

def closed?
  @status_mutex.synchronize { @status == :closed }
end

#closing? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns true if this AMQP 0.9.1 connection is closing.

Returns:

  • (Boolean) —

    true if this AMQP 0.9.1 connection is closing



459
460
461
 
# File 'lib/bunny/session.rb', line 459

def closing?
  @status_mutex.synchronize { @status == :closing }
end

#configure_socket(&block) ⇒ Object

Provides a way to fine tune the socket used by connection. Accepts a block that the socket will be yielded to.

Raises:

  • (ArgumentError) 


306
307
308
309
310
311
312
 
# File 'lib/bunny/session.rb', line 306

def configure_socket(&block)
  raise ArgumentError, "No block provided!" if block.nil?

  @transport_mutex.synchronize do
    @transport.configure_socket(&block)
  end
end

#connecting? ⇒ Boolean

Returns true if this connection is still not fully open.

Returns:

  • (Boolean) —

    true if this connection is still not fully open



453
454
455
 
# File 'lib/bunny/session.rb', line 453

def connecting?
  status == :connecting
end

#create_channel(n = nil, consumer_pool_size = 1, consumer_pool_abort_on_exception = false, consumer_pool_shutdown_timeout = 60) ⇒ Bunny::Channel Also known as: channel

Opens a new channel and returns it. This method will block the calling thread until the response is received and the channel is guaranteed to be opened (this operation is very fast and inexpensive).

Returns:

Raises:

  • (ArgumentError) 


393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
 
# File 'lib/bunny/session.rb', line 393

def create_channel(n = nil, consumer_pool_size = 1, consumer_pool_abort_on_exception = false, consumer_pool_shutdown_timeout = 60)
  raise ArgumentError, "channel number 0 is reserved in the protocol and cannot be used" if 0 == n
  raise ConnectionAlreadyClosed if manually_closed?
  raise RuntimeError, "this connection is not open. Was Bunny::Session#start invoked? Is automatic recovery enabled?" if !connected?

  @channel_mutex.synchronize do
    if n && (ch = @channels[n])
      ch
    else
      work_pool = ConsumerWorkPool.new(consumer_pool_size || 1, consumer_pool_abort_on_exception, consumer_pool_shutdown_timeout)
      ch = Bunny::Channel.new(self, n, {
        work_pool: work_pool
      })
      ch.open
      ch
    end
  end
end

#decrement_recovery_attemp_counter! ⇒ Object 



1069
1070
1071
1072
1073
1074
1075
 
# File 'lib/bunny/session.rb', line 1069

def decrement_recovery_attemp_counter!
  if @recovery_attempts
    @recovery_attempts -= 1
    @logger.debug "#{@recovery_attempts} recovery attempts left"
  end
  @recovery_attempts
end

#delete_recorded_consumer(consumer_tag) ⇒ Object

Parameters:

  • consumer_tag (String) 


989
990
991
 
# File 'lib/bunny/session.rb', line 989

def delete_recorded_consumer(consumer_tag)
  @topology_registry.delete_recorded_consumer(consumer_tag)
end

#delete_recorded_exchange(exchange) ⇒ Object

Parameters:



925
926
927
 
# File 'lib/bunny/session.rb', line 925

def delete_recorded_exchange(exchange)
  @topology_registry.delete_recorded_exchange(exchange)
end

#delete_recorded_exchange_binding(ch, source_name, destination_name, routing_key, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • source_name (String)
  • destination_name (String)
  • routing_key (String)
  • arguments (Hash) 


971
972
973
 
# File 'lib/bunny/session.rb', line 971

def delete_recorded_exchange_binding(ch, source_name, destination_name, routing_key, arguments)
  @topology_registry.delete_recorded_exchange_binding(ch, source_name, destination_name, routing_key, arguments)
end

#delete_recorded_exchange_named(name) ⇒ Object

Parameters:

  • name (String) 


931
932
933
 
# File 'lib/bunny/session.rb', line 931

def delete_recorded_exchange_named(name)
  @topology_registry.delete_recorded_exchange_named(name)
end

#delete_recorded_queue_binding(ch, exchange_name, queue_name, routing_key, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • exchange_name (String)
  • queue_name (String)
  • routing_key (String)
  • arguments (Hash) 


951
952
953
 
# File 'lib/bunny/session.rb', line 951

def delete_recorded_queue_binding(ch, exchange_name, queue_name, routing_key, arguments)
  @topology_registry.delete_recorded_queue_binding(ch, exchange_name, queue_name, routing_key, arguments)
end

#delete_recorded_queue_named(name) ⇒ Object

Parameters:

  • name (String) 


903
904
905
 
# File 'lib/bunny/session.rb', line 903

def delete_recorded_queue_named(name)
  @topology_registry.delete_recorded_queue_named(name)
end

#delete_recoreded_queue(queue) ⇒ Object

Parameters:



897
898
899
 
# File 'lib/bunny/session.rb', line 897

def delete_recoreded_queue(queue)
  @topology_registry.delete_recorded_queue(queue)
end

#exception_occurred? ⇒ Boolean

Returns true if any exceptions have been accumulated by the session error handler. Only works when using the default ExceptionAccumulator handler.

Returns:

  • (Boolean) —

    true if exceptions have been accumulated

Raises:

  • (NoMethodError) —

    if the session error handler doesn't respond to #any?



527
528
529
 
# File 'lib/bunny/session.rb', line 527

def exception_occurred?
  @session_error_handler.any?
end

#exceptions ⇒ Array<Exception>

Returns all accumulated exceptions from the session error handler. Only works when using the default ExceptionAccumulator handler.

Returns:

  • (Array<Exception>) —

    all accumulated exceptions

Raises:

  • (NoMethodError) —

    if the session error handler doesn't respond to #all



537
538
539
 
# File 'lib/bunny/session.rb', line 537

def exceptions
  @session_error_handler.all
end

#exchange_exists?(name) ⇒ Boolean

Checks if a exchange with given name exists.

Implemented using exchange.declare with passive set to true and a one-off (short lived) channel under the hood.

Parameters:

  • name (String) —

    Exchange name

Returns:

  • (Boolean) —

    true if exchange exists



602
603
604
605
606
607
608
609
610
611
612
 
# File 'lib/bunny/session.rb', line 602

def exchange_exists?(name)
  ch = create_channel
  begin
    ch.exchange(name, passive: true)
    true
  rescue Bunny::NotFound => _
    false
  ensure
    ch.close if ch.open?
  end
end

#find_channel(number) ⇒ Object 



680
681
682
 
# File 'lib/bunny/session.rb', line 680

def find_channel(number)
  @channels[number]
end

#give_up_on_recovery ⇒ Object 



1050
1051
1052
1053
1054
1055
1056
 
# File 'lib/bunny/session.rb', line 1050

def give_up_on_recovery
  @logger.error "Ran out of recovery attempts (limit set to #{@max_recovery_attempts}), giving up"
  @transport.close
  self.close(false)
  @manually_closed = false
  notify_of_recovery_attempts_exhausted
end

#handle_frame(ch_number, method) ⇒ Object

Handles incoming frames and dispatches them.

Channel methods (channel.open-ok, channel.close-ok) are handled by the session itself. Connection level errors result in exceptions being raised. Deliveries and other methods are passed on to channels to dispatch.



723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
 
# File 'lib/bunny/session.rb', line 723

def handle_frame(ch_number, method)
  @logger.debug { "Session#handle_frame on #{ch_number}: #{method.inspect}" }
  case method
  when AMQ::Protocol::Channel::OpenOk then
    @continuations.push(method)
  when AMQ::Protocol::Channel::CloseOk then
    @continuations.push(method)
  when AMQ::Protocol::Connection::Close then
    if recover_from_connection_close?
      @logger.warn "Recovering from connection.close (#{method.reply_text})"
      clean_up_on_shutdown
      handle_network_failure(instantiate_connection_level_exception(method))
    else
      clean_up_and_fail_on_connection_close!(method)
    end
  when AMQ::Protocol::Connection::CloseOk then
    @last_connection_close_ok = method
    begin
      @continuations.clear
    rescue StandardError => e
      @logger.error e.class.name
      @logger.error e.message
      @logger.error e.backtrace
    ensure
      @continuations.push(:__unblock__)
    end
  when AMQ::Protocol::Connection::Blocked then
    @blocked = true
    @block_callback.call(method) if @block_callback
  when AMQ::Protocol::Connection::Unblocked then
    @blocked = false
    @unblock_callback.call(method) if @unblock_callback
  when AMQ::Protocol::Connection::UpdateSecretOk then
    @continuations.push(method)
  when AMQ::Protocol::Channel::Close then
    begin
      ch = synchronised_find_channel(ch_number)
      # this includes sending a channel.close-ok and
      # potentially invoking a user-provided callback,
      # avoid doing that while holding a mutex lock. MK.
      ch.handle_method(method)
    ensure
      if ch.nil?
        @logger.warn "Received a server-sent channel.close but the channel was not found locally. Ignoring the frame."
      else
        # synchronises on @channel_mutex under the hood
        self.unregister_channel(ch)
      end
    end
  when AMQ::Protocol::Basic::GetEmpty then
    ch = find_channel(ch_number)
    ch.handle_basic_get_empty(method)
  else
    if ch = find_channel(ch_number)
      ch.handle_method(method)
    else
      @logger.warn "Channel #{ch_number} is not open on this connection!"
    end
  end
end

#handle_frameset(ch_number, frames) ⇒ Object 



790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
 
# File 'lib/bunny/session.rb', line 790

def handle_frameset(ch_number, frames)
  method = frames.first
  ch = find_channel(ch_number)

  unless ch
    @logger.warn "Channel #{ch_number} is not open on this connection!"
    return
  end

  case method
  when AMQ::Protocol::Basic::GetOk    then ch.handle_basic_get_ok(*frames)
  when AMQ::Protocol::Basic::GetEmpty then ch.handle_basic_get_empty(*frames)
  when AMQ::Protocol::Basic::Return   then ch.handle_basic_return(*frames)
  else                                     ch.handle_frameset(*frames)
  end
end

#handle_network_failure(exception) ⇒ Object

Raises:



813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
 
# File 'lib/bunny/session.rb', line 813

def handle_network_failure(exception)
  raise NetworkErrorWrapper.new(exception) unless @threaded

  @status_mutex.synchronize { @status = :disconnected }

  acquired = @recovery_mutex.synchronize do
    if @recovering_from_network_failure
      false
    else
      @recovering_from_network_failure = true
    end
  end
  return unless acquired

  begin
    if recoverable_network_failure?(exception)
      announce_network_failure_recovery
      @channel_mutex.synchronize do
        @channels.each do |n, ch|
          ch.connection_closed!
          ch.maybe_kill_consumer_work_pool!
        end
      end
      @reader_loop.stop if @reader_loop
      maybe_shutdown_heartbeat_sender

      loop do
        break unless recover_connection_and_channels

        recover_topology

        if open?
          mark_channels_after_recovery!
          notify_of_recovery_completion
          break
        end

        if should_retry_recovery?
          decrement_recovery_attemp_counter!
          announce_network_failure_recovery
        else
          give_up_on_recovery
          break
        end
      end
    else
      @logger.error "Exception #{exception.message} is considered unrecoverable..."
    end
  ensure
    @recovery_mutex.synchronize { @recovering_from_network_failure = false }
  end
end

#heartbeat_from(options) ⇒ Object 



1414
1415
1416
 
# File 'lib/bunny/session.rb', line 1414

def heartbeat_from(options)
  options[:heartbeat] || options[:heartbeat_timeout] || options[:requested_heartbeat] || options[:heartbeat_interval] || DEFAULT_HEARTBEAT
end

#heartbeat_interval ⇒ Integer

Deprecated.

Returns Heartbeat timeout (not interval) used.

Returns:

  • (Integer) —

    Heartbeat timeout (not interval) used



267
 
# File 'lib/bunny/session.rb', line 267

def heartbeat_interval; self.heartbeat; end

#heartbeat_timeout ⇒ Integer

Returns Heartbeat timeout used.

Returns:

  • (Integer) —

    Heartbeat timeout used



270
 
# File 'lib/bunny/session.rb', line 270

def heartbeat_timeout; self.heartbeat; end

#host ⇒ Object 



289
290
291
 
# File 'lib/bunny/session.rb', line 289

def host
  @transport ? @transport.host : host_from_address(@addresses[@address_index])
end

#host_from_address(address) ⇒ Object 



1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
 
# File 'lib/bunny/session.rb', line 1355

def host_from_address(address)
  # we need to handle cases such as [2001:db8:85a3:8d3:1319:8a2e:370:7348]:5671
  last_colon                  = address.rindex(":")
  last_closing_square_bracket = address.rindex("]")

  if last_closing_square_bracket.nil?
    parts = address.split(":")
    # this looks like an unquoted IPv6 address, so emit a warning
    if parts.size > 2
      @logger.warn "Address #{address} looks like an unquoted IPv6 address. Make sure you quote IPv6 addresses like so: [2001:db8:85a3:8d3:1319:8a2e:370:7348]"
    end
    return parts[0]
  end

  if last_closing_square_bracket < last_colon
    # there is a port
    address[0, last_colon]
  elsif last_closing_square_bracket > last_colon
    address
  end
end

#host_with_port?(address) ⇒ Boolean

Returns:

  • (Boolean) 


1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
 
# File 'lib/bunny/session.rb', line 1342

def host_with_port?(address)
  # we need to handle cases such as [2001:db8:85a3:8d3:1319:8a2e:370:7348]:5671
  last_colon                  = address.rindex(":")
  last_closing_square_bracket = address.rindex("]")

  if last_closing_square_bracket.nil?
    address.include?(":")
  else
    last_closing_square_bracket < last_colon
  end
end

#hostname ⇒ String

Returns RabbitMQ hostname (or IP address) used.

Returns:

  • (String) —

    RabbitMQ hostname (or IP address) used



257
 
# File 'lib/bunny/session.rb', line 257

def hostname;     self.host;  end

#inspect ⇒ Object 



1592
1593
1594
 
# File 'lib/bunny/session.rb', line 1592

def inspect
  to_s
end

#instantiate_connection_level_exception(frame) ⇒ Object 



1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
 
# File 'lib/bunny/session.rb', line 1261

def instantiate_connection_level_exception(frame)
  case frame
  when AMQ::Protocol::Connection::Close then
    klass = case frame.reply_code
            when 320 then
              ConnectionForced
            when 501 then
              FrameError
            when 503 then
              CommandInvalid
            when 504 then
              ChannelError
            when 505 then
              UnexpectedFrame
            when 506 then
              ResourceError
            when 530 then
              NotAllowedError
            when 541 then
              InternalError
            else
              raise "Unknown reply code: #{frame.reply_code}, text: #{frame.reply_text}"
            end

    klass.new("Connection-level error: #{frame.reply_text}", self, frame)
  end
end

#local_port ⇒ Integer

Returns Client socket port.

Returns:

  • (Integer) —

    Client socket port



315
316
317
 
# File 'lib/bunny/session.rb', line 315

def local_port
  @transport.local_address.ip_port
end

#manually_closed? ⇒ Boolean

Returns true if this AMQP 0.9.1 connection has been closed by the user (as opposed to the server).

Returns:

  • (Boolean) —

    true if this AMQP 0.9.1 connection has been closed by the user (as opposed to the server)



469
470
471
 
# File 'lib/bunny/session.rb', line 469

def manually_closed?
  @status_mutex.synchronize { @manually_closed == true }
end

#mark_channels_after_recovery! ⇒ Object 



1094
1095
1096
1097
1098
1099
1100
 
# File 'lib/bunny/session.rb', line 1094

def mark_channels_after_recovery!
  @channel_mutex.synchronize do
    @channels.each do |n, ch|
      ch.recovery_completed!
    end
  end
end

#maybe_shutdown_reader_loop ⇒ Object 



1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
 
# File 'lib/bunny/session.rb', line 1456

def maybe_shutdown_reader_loop
  if @reader_loop
    @reader_loop.stop
    if threaded?
      # this is the easiest way to wait until the loop
      # is guaranteed to have terminated
      @reader_loop.terminate_with(ShutdownSignal)
      @reader_loop.join
    else
      # single threaded mode, nothing to do. MK.
    end
  end

  @reader_loop = nil
end

#next_channel_id ⇒ Object 



1419
1420
1421
 
# File 'lib/bunny/session.rb', line 1419

def next_channel_id
  @channel_id_allocator.next_channel_id
end

#notify_of_recovery_attempt_start ⇒ Object 



1246
1247
1248
 
# File 'lib/bunny/session.rb', line 1246

def notify_of_recovery_attempt_start
  @recovery_attempt_started.call if @recovery_attempt_started
end

#notify_of_recovery_attempts_exhausted ⇒ Object 



1256
1257
1258
 
# File 'lib/bunny/session.rb', line 1256

def notify_of_recovery_attempts_exhausted
  @recovery_attempts_exhausted.call if @recovery_attempts_exhausted
end

#notify_of_recovery_completion ⇒ Object 



1251
1252
1253
 
# File 'lib/bunny/session.rb', line 1251

def notify_of_recovery_completion
  @recovery_completed.call if @recovery_completed
end

#on_blocked {|AMQ::Protocol::Connection::Blocked| ... } ⇒ Object

Defines a callback that will be executed when RabbitMQ blocks the connection because it is running low on memory or disk space (as configured via config file and/or rabbitmqctl).

Yields:

  • (AMQ::Protocol::Connection::Blocked) —

    connection.blocked method which provides a reason for blocking



493
494
495
 
# File 'lib/bunny/session.rb', line 493

def on_blocked(&block)
  @block_callback = block
end

#on_unblocked(&block) ⇒ Object

Defines a callback that will be executed when RabbitMQ unblocks the connection that was previously blocked, e.g. because the memory or disk space alarm has cleared.

See Also:



502
503
504
 
# File 'lib/bunny/session.rb', line 502

def on_unblocked(&block)
  @unblock_callback = block
end

#open? ⇒ Boolean Also known as: connected?

Returns true if this AMQP 0.9.1 connection is open.

Returns:

  • (Boolean) —

    true if this AMQP 0.9.1 connection is open



474
475
476
477
478
 
# File 'lib/bunny/session.rb', line 474

def open?
  @status_mutex.synchronize do
    (status == :open || status == :connected || status == :connecting) && @transport.open?
  end
end

#open_channel(ch) ⇒ Object 



649
650
651
652
653
654
655
656
657
658
659
660
661
662
 
# File 'lib/bunny/session.rb', line 649

def open_channel(ch)
  @channel_mutex.synchronize do
    n = ch.number
    self.register_channel(ch)

    @transport_mutex.synchronize do
      @transport.send_frame(AMQ::Protocol::Channel::Open.encode(n, AMQ::Protocol::EMPTY_STRING))
    end
    @last_channel_open_ok = wait_on_continuations
    raise_if_continuation_resulted_in_a_connection_error!

    @last_channel_open_ok
  end
end

#password ⇒ String

Returns Password used.

Returns:

  • (String) —

    Password used



261
 
# File 'lib/bunny/session.rb', line 261

def password;     self.pass;  end

#password_from(options) ⇒ Object 



1409
1410
1411
 
# File 'lib/bunny/session.rb', line 1409

def password_from(options)
  options[:password] || options[:pass] || options[:pwd] || DEFAULT_PASSWORD
end

#port ⇒ Object 



293
294
295
 
# File 'lib/bunny/session.rb', line 293

def port
  @transport ? @transport.port : port_from_address(@addresses[@address_index])
end

#port_from(options) ⇒ Object 



1331
1332
1333
1334
1335
1336
1337
1338
1339
 
# File 'lib/bunny/session.rb', line 1331

def port_from(options)
  fallback = if options[:tls] || options[:ssl]
               AMQ::Protocol::TLS_PORT
             else
               AMQ::Protocol::DEFAULT_PORT
             end

  options.fetch(:port, fallback)
end

#port_from_address(address) ⇒ Object 



1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
 
# File 'lib/bunny/session.rb', line 1378

def port_from_address(address)
  # we need to handle cases such as [2001:db8:85a3:8d3:1319:8a2e:370:7348]:5671
  last_colon                  = address.rindex(":")
  last_closing_square_bracket = address.rindex("]")

  if last_closing_square_bracket.nil?
    parts = address.split(":")
    # this looks like an unquoted IPv6 address, so emit a warning
    if parts.size > 2
      @logger.warn "Address #{address} looks like an unquoted IPv6 address. Make sure you quote IPv6 addresses like so: [2001:db8:85a3:8d3:1319:8a2e:370:7348]"
    end
    return parts[1].to_i
  end

  if last_closing_square_bracket < last_colon
    # there is a port
    address[(last_colon + 1)..-1].to_i
  end
end

#queue_exists?(name) ⇒ Boolean

Checks if a queue with given name exists.

Implemented using queue.declare with passive set to true and a one-off (short lived) channel under the hood.

Parameters:

  • name (String) —

    Queue name

Returns:

  • (Boolean) —

    true if queue exists



580
581
582
583
584
585
586
587
588
589
590
591
592
 
# File 'lib/bunny/session.rb', line 580

def queue_exists?(name)
  ch = create_channel
  begin
    ch.queue(name, passive: true)
    true
  rescue Bunny::ResourceLocked => _
    true
  rescue Bunny::NotFound => _
    false
  ensure
    ch.close if ch.open?
  end
end

#raise_if_continuation_resulted_in_a_connection_error! ⇒ Object

Raises:

  • (@last_connection_error) 


785
786
787
 
# File 'lib/bunny/session.rb', line 785

def raise_if_continuation_resulted_in_a_connection_error!
  raise @last_connection_error if @last_connection_error
end

#raise_on_exception! ⇒ Object

Raises the first accumulated exception if any exist. Only works when using the default ExceptionAccumulator handler.

This is useful for checking errors at safe points in your application, such as after completing a batch of operations.

Raises:

  • (Exception) —

    the first accumulated exception

  • (NoMethodError) —

    if the session error handler doesn't respond to #raise_first!



560
561
562
 
# File 'lib/bunny/session.rb', line 560

def raise_on_exception!
  @session_error_handler.raise_first!
end

#reader_loop ⇒ Object 



1451
1452
1453
 
# File 'lib/bunny/session.rb', line 1451

def reader_loop
  @reader_loop ||= ReaderLoop.new(@transport, self, @session_error_handler)
end

#record_consumer_with(ch, consumer_tag, queue_name, callable, manual_ack, exclusive, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • consumer_tag (String)
  • queue_name (String)
  • callable (#call)
  • manual_ack (Boolean)
  • exclusive (Boolean)
  • arguments (Hash) 


983
984
985
 
# File 'lib/bunny/session.rb', line 983

def record_consumer_with(ch, consumer_tag, queue_name, callable, manual_ack, exclusive, arguments)
  @topology_registry.record_consumer_with(ch, consumer_tag, queue_name, callable, manual_ack, exclusive, arguments)
end

#record_exchange(exchange) ⇒ Object

Parameters:



909
910
911
 
# File 'lib/bunny/session.rb', line 909

def record_exchange(exchange)
  @topology_registry.record_exchange(exchange)
end

#record_exchange_binding_with(ch, source_name, destination_name, routing_key, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • source_name (String)
  • destination_name (String)
  • routing_key (String)
  • arguments (Hash) 


961
962
963
 
# File 'lib/bunny/session.rb', line 961

def record_exchange_binding_with(ch, source_name, destination_name, routing_key, arguments)
  @topology_registry.record_exchange_binding_with(ch, source_name, destination_name, routing_key, arguments)
end

#record_exchange_with(ch, name, type, durable, auto_delete, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • name (String)
  • type (String)
  • durable (Boolean)
  • auto_delete (Boolean)
  • arguments (Hash) 


919
920
921
 
# File 'lib/bunny/session.rb', line 919

def record_exchange_with(ch, name, type, durable, auto_delete, arguments)
  @topology_registry.record_exchange_with(ch, name, type, durable, auto_delete, arguments)
end

#record_queue(queue) ⇒ Object

Parameters:



880
881
882
 
# File 'lib/bunny/session.rb', line 880

def record_queue(queue)
  @topology_registry.record_queue(queue)
end

#record_queue_binding_with(ch, exchange_name, queue_name, routing_key, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • exchange_name (String)
  • queue_name (String)
  • routing_key (String)
  • arguments (Hash) 


941
942
943
 
# File 'lib/bunny/session.rb', line 941

def record_queue_binding_with(ch, exchange_name, queue_name, routing_key, arguments)
  @topology_registry.record_queue_binding_with(ch, exchange_name, queue_name, routing_key, arguments)
end

#record_queue_name_change(old_name, new_name) ⇒ Object

Parameters:

  • old_name (String)
  • new_name (String) 


1206
1207
1208
 
# File 'lib/bunny/session.rb', line 1206

def record_queue_name_change(old_name, new_name)
  @topology_registry.record_queue_name_change(old_name, new_name)
end

#record_queue_with(ch, name, server_named, durable, auto_delete, exclusive, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • name (String)
  • server_named (Boolean)
  • durable (Boolean)
  • auto_delete (Boolean)
  • exclusive (Boolean)
  • arguments (Hash) 


891
892
893
 
# File 'lib/bunny/session.rb', line 891

def record_queue_with(ch, name, server_named, durable, auto_delete, exclusive, arguments)
  @topology_registry.record_queue_with(ch, name, server_named, durable, auto_delete, exclusive, arguments)
end

#recover_channel_topology(ch) ⇒ Object

Recovers topology (exchanges, queues, bindings, consumers) for a single channel. Intended for use after Channel#reopen to restore the channel to a working state with its original consumers.

Parameters:



639
640
641
642
 
# File 'lib/bunny/session.rb', line 639

def recover_channel_topology(ch)
  filter = Proc.new { |entity| entity.channel == ch }
  recover_topology_with(filter)
end

#recover_channels ⇒ Object 



1083
1084
1085
1086
1087
1088
1089
1090
1091
 
# File 'lib/bunny/session.rb', line 1083

def recover_channels
  @channel_mutex.synchronize do
    @channels.each do |n, ch|
      ch.open
      ch.recovering!
      ch.recover_from_network_failure
    end
  end
end

#recover_connection_and_channels ⇒ Object 



1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
 
# File 'lib/bunny/session.rb', line 1005

def recover_connection_and_channels
  sleep @network_recovery_interval
  @logger.debug "Will attempt connection recovery..."
  notify_of_recovery_attempt_start

  self.initialize_transport

  @logger.debug "Retrying connection on next host in line: #{@transport.host}:#{@transport.port}"
  self.start

  if open?
    @logger.debug "Connection is now open"
    if @reset_recovery_attempt_counter_after_reconnection
      @logger.debug "Resetting recovery attempt counter after successful reconnection"
      reset_recovery_attempt_counter!
    else
      @logger.debug "Not resetting recovery attempt counter after successful reconnection, as configured"
    end

    recover_channels
    true
  else
    raise TCPConnectionFailed.new("connection is not open after a connection recovery attempt", @transport.host, @transport.port)
  end
rescue HostListDepleted
  reset_address_index
  retry
rescue => e
  if recoverable_network_failure?(e)
    @logger.warn "TCP connection failed"
    if should_retry_recovery?
      @logger.warn "Reconnecting in #{@network_recovery_interval} seconds"
      decrement_recovery_attemp_counter!
      announce_network_failure_recovery
      retry
    else
      give_up_on_recovery
      false
    end
  else
    raise e
  end
end

#recover_consumer(c) ⇒ Object

Parameters:



1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
 
# File 'lib/bunny/session.rb', line 1234

def recover_consumer(c)
  c.channel.maybe_reinitialize_consumer_pool!
  if c.callable.is_a?(Bunny::Consumer)
    c.channel.basic_consume_with(c.callable)
  else
    c.channel.basic_consume(c.queue_name, c.consumer_tag, !c.manual_ack, c.exclusive, c.arguments) do |*args|
      c.callable.call(*args)
    end
  end
end

#recover_exchange(x) ⇒ Object

Parameters:



1171
1172
1173
1174
1175
1176
1177
1178
 
# File 'lib/bunny/session.rb', line 1171

def recover_exchange(x)
  opts = {
    durable: x.durable,
    auto_delete: x.auto_delete,
    arguments: x.arguments
  }
  x.channel.exchange_declare(x.name, x.type, opts)
end

#recover_exchange_binding(rb) ⇒ Object

Parameters:

  • rb (Bunny::RecordedExchangeBindingBinding) 


1223
1224
1225
1226
1227
1228
1229
1230
 
# File 'lib/bunny/session.rb', line 1223

def recover_exchange_binding(rb)
  opts = {
    routing_key: rb.routing_key,
    arguments: rb.arguments
  }

  rb.channel.exchange_bind_without_recording_topology(rb.source, rb.destination, opts)
end

#recover_from_connection_close? ⇒ Boolean

Returns:

  • (Boolean) 


808
809
810
 
# File 'lib/bunny/session.rb', line 808

def recover_from_connection_close?
  @recover_from_connection_close
end

#recover_queue(q) ⇒ Object

Parameters:



1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
 
# File 'lib/bunny/session.rb', line 1182

def recover_queue(q)
  opts = {
    durable: q.durable,
    auto_delete: q.auto_delete,
    exclusive: q.exclusive,
    arguments: q.arguments
  }

  old_name = q.name
  # this response carries the server-generated name
  queue_declare_ok = q.channel.queue_declare(q.name_to_use_for_recovery, opts)
  new_name = queue_declare_ok.queue

  # if the name has changed, update all the bindings where
  # this queue is the destination, then all consumers
  if new_name != old_name
    record_queue_name_change(old_name, new_name)
    q.channel.record_queue_name_change(old_name, new_name)
  end
end

#recover_queue_binding(rb) ⇒ Object

Parameters:



1212
1213
1214
1215
1216
1217
1218
1219
 
# File 'lib/bunny/session.rb', line 1212

def recover_queue_binding(rb)
  opts = {
    routing_key: rb.routing_key,
    arguments: rb.arguments
  }

  rb.channel.queue_bind_without_recording_topology(rb.destination, rb.source, opts)
end

#recover_topology ⇒ Object 



1103
1104
1105
1106
 
# File 'lib/bunny/session.rb', line 1103

def recover_topology
  @logger.debug "Will recover topology now"
  recover_topology_with
end

#recover_topology_with(filter = nil) ⇒ Object 



1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
 
# File 'lib/bunny/session.rb', line 1109

def recover_topology_with(filter = nil)
  exchanges = @topology_registry.filtered_exchanges.reject(&:predeclared?)
  queues = @topology_registry.filtered_queues
  queue_bindings = @topology_registry.filtered_queue_bindings
  exchange_bindings = @topology_registry.filtered_exchange_bindings
  consumers = @topology_registry.filtered_consumers

  if filter
    exchanges = exchanges.select(&filter)
    queues = queues.select(&filter)
    queue_bindings = queue_bindings.select(&filter)
    exchange_bindings = exchange_bindings.select(&filter)
    consumers = consumers.select(&filter)
  end

  @logger.debug { "Will recover #{exchanges.size} exchange(s)" }
  exchanges.each do |x|
    begin
      recover_exchange(x)
    rescue ::StandardError => e
      @logger.error "Caught an exception while recovering exchange #{x.name}: #{e.inspect}"
    end
  end

  @logger.debug { "Will recover #{queues.size} queue(s)" }
  queues.each do |q|
    begin
      recover_queue(q)
    rescue ::StandardError => e
      @logger.error "Caught an exception while recovering queue #{q.name}: #{e.inspect}"
    end
  end

  @logger.debug { "Will recover #{queue_bindings.size + exchange_bindings.size} binding(s)" }
  queue_bindings.each do |b|
    begin
      recover_queue_binding(b)
    rescue ::StandardError => e
      @logger.error "Caught an exception while recovering a binding of queue #{b.destination}: #{e.inspect}"
    end
  end

  exchange_bindings.each do |b|
    begin
      recover_exchange_binding(b)
    rescue ::StandardError => e
      @logger.error "Caught an exception while recovering a binding of exchange #{b.source}: #{e.inspect}"
    end
  end

  @logger.debug { "Will recover #{consumers.size} consumer(s)" }
  consumers.each do |c|
    begin
      recover_consumer(c)
    rescue ::StandardError => e
      @logger.error "Caught an exception while recovering consumer #{c.consumer_tag} on queue #{c.queue_name}: #{e.inspect}"
    end
  end
end

#recoverable_network_failure?(exception) ⇒ Boolean

Returns:

  • (Boolean) 


868
869
870
 
# File 'lib/bunny/session.rb', line 868

def recoverable_network_failure?(exception)
  @recoverable_exceptions.any? {|x| exception.kind_of? x}
end

#recovering_from_network_failure? ⇒ Boolean

Returns:

  • (Boolean) 


874
875
876
 
# File 'lib/bunny/session.rb', line 874

def recovering_from_network_failure?
  @recovery_mutex.synchronize { @recovering_from_network_failure }
end

#recovery_attempts_limited? ⇒ Boolean

Returns:

  • (Boolean) 


1059
1060
1061
 
# File 'lib/bunny/session.rb', line 1059

def recovery_attempts_limited?
  !!@max_recovery_attempts
end

#register_channel(ch) ⇒ Object 



1429
1430
1431
1432
1433
 
# File 'lib/bunny/session.rb', line 1429

def register_channel(ch)
  @channel_mutex.synchronize do
    @channels[ch.number] = ch
  end
end

#release_channel_id(i) ⇒ Object 



1424
1425
1426
 
# File 'lib/bunny/session.rb', line 1424

def release_channel_id(i)
  @channel_id_allocator.release_channel_id(i)
end

#reset_address_index ⇒ Object 



297
298
299
 
# File 'lib/bunny/session.rb', line 297

def reset_address_index
  @address_index_mutex.synchronize { @address_index = 0 }
end

#reset_recovery_attempt_counter! ⇒ Object 



1078
1079
1080
 
# File 'lib/bunny/session.rb', line 1078

def reset_recovery_attempt_counter!
  @recovery_attempts = @max_recovery_attempts
end

#send_frame(frame, signal_activity = true) ⇒ Object

Sends frame to the peer, checking that connection is open. Exposed primarily for Bunny::Channel

Raises:



1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
 
# File 'lib/bunny/session.rb', line 1495

def send_frame(frame, signal_activity = true)
  if open?
    # @transport_mutex.synchronize do
    #   @transport.write(frame.encode)
    # end
    @transport.write(frame.encode)
    signal_activity! if signal_activity
  else
    raise ConnectionClosedError.new(frame)
  end
end

#send_frame_without_timeout(frame, signal_activity = true) ⇒ Object

Sends frame to the peer, checking that connection is open. Uses transport implementation that does not perform timeout control. Exposed primarily for Bunny::Channel.

Raises:



1513
1514
1515
1516
1517
1518
1519
1520
 
# File 'lib/bunny/session.rb', line 1513

def send_frame_without_timeout(frame, signal_activity = true)
  if open?
    @transport.write_without_timeout(frame.encode)
    signal_activity! if signal_activity
  else
    raise ConnectionClosedError.new(frame)
  end
end

#send_frameset(frames, channel) ⇒ Object

Sends multiple frames, in one go. For thread safety this method takes a channel object and synchronizes on it.



1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
 
# File 'lib/bunny/session.rb', line 1526

def send_frameset(frames, channel)
  # some developers end up sharing channels between threads and when multiple
  # threads publish on the same channel aggressively, at some point frames will be
  # delivered out of order and broker will raise 505 UNEXPECTED_FRAME exception.
  # If we synchronize on the channel, however, this is both thread safe and pretty fine-grained
  # locking. Note that "single frame" methods technically do not need this kind of synchronization
  # (no incorrect frame interleaving of the same kind as with basic.publish isn't possible) but we
  # still recommend not sharing channels between threads except for consumer-only cases in the docs. MK.
  channel.synchronize do
    # see rabbitmq/rabbitmq-server#156
    if open?
      data = frames.reduce(+"") { |acc, frame| acc << frame.encode }
      @transport.write(data)
      signal_activity!
    else
      raise ConnectionClosedError.new(frames)
    end
  end
end

#send_frameset_without_timeout(frames, channel) ⇒ Object

Sends multiple frames, one by one. For thread safety this method takes a channel object and synchronizes on it. Uses transport implementation that does not perform timeout control.



1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
 
# File 'lib/bunny/session.rb', line 1551

def send_frameset_without_timeout(frames, channel)
  # some developers end up sharing channels between threads and when multiple
  # threads publish on the same channel aggressively, at some point frames will be
  # delivered out of order and broker will raise 505 UNEXPECTED_FRAME exception.
  # If we synchronize on the channel, however, this is both thread safe and pretty fine-grained
  # locking. See a note about "single frame" methods in a comment in `send_frameset`. MK.
  channel.synchronize do
    if open?
      data = frames.reduce(+"") { |acc, frame| acc << frame.encode }
      @transport.write_without_timeout(data)
      signal_activity!
    else
      raise ConnectionClosedError.new(frames)
    end
  end
end

#send_raw_without_timeout(data, channel) ⇒ Object 



1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
 
# File 'lib/bunny/session.rb', line 1569

def send_raw_without_timeout(data, channel)
  # some developers end up sharing channels between threads and when multiple
  # threads publish on the same channel aggressively, at some point frames will be
  # delivered out of order and broker will raise 505 UNEXPECTED_FRAME exception.
  # If we synchronize on the channel, however, this is both thread safe and pretty fine-grained
  # locking. Note that "single frame" methods do not need this kind of synchronization. MK.
  channel.synchronize do
    if open?
      @transport.write(data)
      signal_activity!
    else
      raise ConnectionClosedError.new("pre-encoded data (#{data.bytesize} bytes)")
    end
  end
end

#should_retry_recovery? ⇒ Boolean

Returns:

  • (Boolean) 


1064
1065
1066
 
# File 'lib/bunny/session.rb', line 1064

def should_retry_recovery?
  !recovery_attempts_limited? || @recovery_attempts > 1
end

#signal_activity! ⇒ Object 



1485
1486
1487
 
# File 'lib/bunny/session.rb', line 1485

def signal_activity!
  @heartbeat_sender.signal_activity! if @heartbeat_sender
end

#start ⇒ Object

Starts the connection process.

See Also:



324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
 
# File 'lib/bunny/session.rb', line 324

def start
  return self if connected?

  @status_mutex.synchronize { @status = :connecting }
  # reset here for cases when automatic network recovery kicks in
  # when we were blocked. MK.
  @blocked       = false
  self.reset_continuations

  begin
    begin
      # close existing transport if we have one,
      # to not leak sockets
      @transport_mutex.synchronize do
        @transport.maybe_initialize_socket
        @transport.post_initialize_socket
        @transport.connect
      end

      if @transport.read_timeout.nil?
        @transport.read_timeout = @transport.connect_timeout || Transport::DEFAULT_READ_TIMEOUT
      end

      self.init_connection
      self.open_connection

      @reader_loop = nil
      self.start_reader_loop if threaded?

    rescue TCPConnectionFailed => e
      @logger.warn e.message
      self.initialize_transport
      @logger.warn "Will try to connect to the next endpoint in line: #{@transport.host}:#{@transport.port}"

      return self.start
    rescue
      @status_mutex.synchronize { @status = :not_connected }
      raise
    end
  rescue HostListDepleted
    self.reset_address_index
    @status_mutex.synchronize { @status = :not_connected }
    raise TCPConnectionFailedForAllHosts
  end
  @status_mutex.synchronize { @manually_closed = false }

  self
end

#start_reader_loop ⇒ Object 



1446
1447
1448
 
# File 'lib/bunny/session.rb', line 1446

def start_reader_loop
  reader_loop.start
end

#synchronised_find_channel(number) ⇒ Object 



685
686
687
 
# File 'lib/bunny/session.rb', line 685

def synchronised_find_channel(number)
  @channel_mutex.synchronize { @channels[number] }
end

#threaded? ⇒ Boolean

Returns true if this connection uses a separate thread for I/O activity.

Returns:

  • (Boolean) —

    true if this connection uses a separate thread for I/O activity



285
286
287
 
# File 'lib/bunny/session.rb', line 285

def threaded?
  @threaded
end

#to_s ⇒ String

Returns:

  • (String) 


1587
1588
1589
1590
 
# File 'lib/bunny/session.rb', line 1587

def to_s
  oid = ("0x%x" % (self.object_id << 1))
  "#<#{self.class.name}:#{oid} #{@user}@#{host}:#{port}, vhost=#{@vhost}, addresses=[#{@addresses.join(',')}]>"
end

#transport_write_timeout ⇒ Float

Socket operation write timeout used by this connection

Returns:

  • (Float) 


384
385
386
 
# File 'lib/bunny/session.rb', line 384

def transport_write_timeout
  @transport.write_timeout
end

#unregister_channel(ch) ⇒ Object 



1436
1437
1438
1439
1440
1441
1442
1443
 
# File 'lib/bunny/session.rb', line 1436

def unregister_channel(ch)
  @channel_mutex.synchronize do
    n = ch.number

    self.release_channel_id(n)
    @channels.delete(ch.number)
  end
end

#update_secret(value, reason) ⇒ Object 



373
374
375
376
377
378
379
 
# File 'lib/bunny/session.rb', line 373

def update_secret(value, reason)
  @transport.send_frame(AMQ::Protocol::Connection::UpdateSecret.encode(value, reason))
  @last_update_secret_ok = wait_on_continuations
  raise_if_continuation_resulted_in_a_connection_error!

  @last_update_secret_ok
end

#username ⇒ String

Returns Username used.

Returns:

  • (String) —

    Username used



259
 
# File 'lib/bunny/session.rb', line 259

def username;     self.user;  end

#username_from(options) ⇒ Object 



1404
1405
1406
 
# File 'lib/bunny/session.rb', line 1404

def username_from(options)
  options[:username] || options[:user] || DEFAULT_USER
end

#uses_ssl? ⇒ Boolean Also known as: ssl?

Returns true if this connection uses TLS (SSL).

Returns:

  • (Boolean) —

    true if this connection uses TLS (SSL)



279
280
281
 
# File 'lib/bunny/session.rb', line 279

def uses_ssl?
  @transport.uses_ssl?
end

#uses_tls? ⇒ Boolean Also known as: tls?

Returns true if this connection uses TLS (SSL).

Returns:

  • (Boolean) —

    true if this connection uses TLS (SSL)



273
274
275
 
# File 'lib/bunny/session.rb', line 273

def uses_tls?
  @transport.uses_tls?
end

#validate_connection_options(options) ⇒ Object 



246
247
248
249
250
251
252
253
254
 
# File 'lib/bunny/session.rb', line 246

def validate_connection_options(options)
  if options[:hosts] && options[:addresses]
    raise ArgumentError, "Connection options can't contain hosts and addresses at the same time"
  end

  if (options[:host] || options[:hostname]) && (options[:hosts] || options[:addresses])
    @logger.warn "Connection options contain both a host and an array of hosts (addresses), please pick one."
  end
end

#vhost_from(options) ⇒ Object 



1399
1400
1401
 
# File 'lib/bunny/session.rb', line 1399

def vhost_from(options)
  options[:virtual_host] || options[:vhost] || DEFAULT_VHOST
end

#virtual_host ⇒ String

Returns Virtual host used.

Returns:

  • (String) —

    Virtual host used



263
 
# File 'lib/bunny/session.rb', line 263

def virtual_host; self.vhost; end

#with_channel(n = nil) ⇒ Bunny::Session

Creates a temporary channel, yields it to the block given to this method and closes it.

Returns:



441
442
443
444
445
446
447
448
449
450
 
# File 'lib/bunny/session.rb', line 441

def with_channel(n = nil)
  ch = create_channel(n)
  begin
    yield ch
  ensure
    ch.close if ch.open?
  end

  self
end