Exception: AceMQ::AMQP::DrainTimeout

Inherits:
StandardError
  • Object
show all
Defined in:
lib/acemq/amqp/connection.rb

Overview

A shutdown that ran out of time with handlers still working.

Raised by Connection#close after every consumer has been stopped and the socket has been closed, because the shutdown still happened — what this says is that it did not finish. The deliveries those handlers were holding were never acknowledged, so the broker will give them to whoever is still consuming the queue: the work may be done twice, which is the ordinary at-least-once case, and nothing is lost.

It is raised rather than logged because the alternative is a drain that reports success having abandoned work in flight, and an operator who has never seen this error has no way to know their grace period is too short.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(stranded, timeout) ⇒ DrainTimeout

Returns a new instance of DrainTimeout.



74
75
76
77
78
79
80
81
82
 
# File 'lib/acemq/amqp/connection.rb', line 74

def initialize(stranded, timeout)
  @stranded = stranded
  @timeout = timeout
  listed = stranded.map { |queue, count| "#{queue} (#{count})" }.join(", ")
  total = stranded.values.sum
  super("the drain did not finish within #{timeout}s: #{total} " \
        "#{total == 1 ? "delivery was" : "deliveries were"} left unsettled " \
        "and will be redelivered — #{listed}")
end

Instance Attribute Details

#stranded ⇒ Hash{String => Integer} (readonly)

How many deliveries were still in flight when time ran out, by queue.

Returns:

  • (Hash{String => Integer}) 


69
70
71
 
# File 'lib/acemq/amqp/connection.rb', line 69

def stranded
  @stranded
end

#timeout ⇒ Object (readonly)

The deadline that expired, in seconds.



72
73
74
 
# File 'lib/acemq/amqp/connection.rb', line 72

def timeout
  @timeout
end