Module: AceMQ::AMQP::Drain

Defined in:
lib/acemq/amqp/connection.rb

Overview

Stopping several consumers against one deadline.

The one place the shared-deadline rule is written down, because Connection#close and Patterns::ConsumerGroup#close both need it and two copies would be two answers to "how long may a shutdown take".

Class Method Summary collapse

Class Method Details

.of(consumers, timeout) ⇒ Array(StandardError, nil, Hash{String => Integer})

Stops every consumer, and says what was left when time ran out.

All of them share one deadline: nothing starts a fresh wait of its own. That is the whole difference between a drain a process can be held to and one that is however many consumers there happen to be times thirty seconds.

Every consumer is stopped even when one of them refuses, and the caller closes the connection either way. Stopping at the first failure would leave the rest of them running and the socket open, so a shutdown that went slightly wrong would become a process that will not exit — which is a worse problem than whatever the first consumer objected to. A consumer that raised is not asked what it is holding: the exception is the thing worth reporting about it, and a second one raised while asking would bury the first.

Three passes rather than one cancel per consumer: every subscription is stopped first, then the handlers of all of them are waited for against the one deadline, then every subscription is released. Cancelling one consumer at a time left the later ones subscribed — and still being handed work that the deadline would then strand — while the earlier ones were being waited for.

Parameters:

  • consumers (Array<Consumer>)
  • timeout (Numeric) —

    seconds for all of them together

Returns:

  • (Array(StandardError, nil, Hash{String => Integer})) —

    the first failure, and what each queue still had in flight when time ran out



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/acemq/amqp/connection.rb', line 118

def self.of(consumers, timeout)
  deadline = now + timeout
  failures = []
  stopped = consumers.select { |consumer| attempt(failures) { consumer.stop } }
  # Polled rather than signalled. A condition variable would be tidier and
  # would mean holding a lock across a handler that may be waiting on a
  # database, which is how a shutdown turns into a deadlock.
  sleep(0.01) while stopped.any? { |c| c.in_flight.positive? } && now < deadline
  stranded = Hash.new(0)
  stopped.each do |consumer|
    # What was still running when the subscription was let go is what the
    # deadline did not wait for, and the number an operator needs: those
    # deliveries are unacknowledged and will come round again.
    attempt(failures) { stranded[consumer.queue] += consumer.release }
  end
  [failures.first, stranded.reject { |_, count| count.zero? }]
end