Class: AceMQ::AMQP::RecoverLiveChannelsOnly Private

Inherits:

Object

• Object
• AceMQ::AMQP::RecoverLiveChannelsOnly

Defined in:

lib/acemq/amqp/transport.rb

Overview

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

Keeps bunny's topology recovery off channels this library has closed.

bunny 3.x records every declaration for replay when a connection comes back, and records it against the channel it was made on. This library declares queues, exchanges and bindings on a short-lived channel of their own and closes it — deliberately, because a refused declaration kills the channel it was made on, and sharing one would take every later declaration down with the first PRECONDITION failure. Recovery then tried to replay those declarations on a channel that was closed minutes earlier, and bunny logged one of these per entity, per recovery:

Caught an exception while recovering exchange acemq.dlx:
#<Bunny::ChannelAlreadyClosed: cannot use a closed channel! Channel id: 1>

It was not only noise. Every failed replay left a consumer work pool behind, so a client that reconnected often grew threads until it was unhealthy: a standing load reached ~54 threads by its 12th forced recovery, against 9 for the same shape built on bunny alone.

So entities whose channel has gone are dropped, and everything on a live channel — the subscription's channel, and the consumers on it — is recovered as bunny intends.

What this gives up. A queue that only existed because of the connection, an exclusive or auto_delete one declared through Transport#declare_queue, is not re-declared on recovery. Durable queues, which is everything this library creates by default, live on the broker and need no replay. Nothing is lost against bunny 2.x either, which has no topology recorder at all and replays nothing.

Duck-typed rather than a Bunny::TopologyRecoveryFilter subclass, because bunny is loaded lazily: naming the constant here would require it at load time and drag a broker client into a process that only reads envelopes.

Instance Method Summary

• #filter_consumers(consumers) ⇒ Object private

  Consumers are never filtered, and filtering them was a far worse bug than the one this class fixes.

• #filter_exchange_bindings(bindings) ⇒ Object private
• #filter_exchanges(exchanges) ⇒ Object private
• #filter_queue_bindings(bindings) ⇒ Object private
• #filter_queues(queues) ⇒ Object private

Instance Method Details

#filter_consumers(consumers) ⇒ Object

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.

Consumers are never filtered, and filtering them was a far worse bug than the one this class fixes. A consumer's channel can read as closed at the moment the filter runs, so asking the same question of it dropped the subscription from recovery: the first version of this took the thread count from 52 to 4 and stopped the client consuming, which is the failure this whole library exists to prevent. Deliveries stopped at 715 while publishing climbed past 17,000.

It is also unnecessary. The leak came from replaying declarations recorded on the short-lived channels this library closes; a consumer is recorded on the subscription's own channel, which stays open for the life of the subscription.

# File 'lib/acemq/amqp/transport.rb', line 339

def filter_consumers(consumers) = consumers

#filter_exchange_bindings(bindings) ⇒ Object

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.

# File 'lib/acemq/amqp/transport.rb', line 326

def filter_exchange_bindings(bindings) = live(bindings)

#filter_exchanges(exchanges) ⇒ Object

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.

# File 'lib/acemq/amqp/transport.rb', line 324

def filter_exchanges(exchanges) = live(exchanges)

#filter_queue_bindings(bindings) ⇒ Object

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.

# File 'lib/acemq/amqp/transport.rb', line 327

def filter_queue_bindings(bindings) = live(bindings)

#filter_queues(queues) ⇒ Object

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.

# File 'lib/acemq/amqp/transport.rb', line 325

def filter_queues(queues) = live(queues)