Class: AceMQ::AMQP::ConfirmWait Private

Inherits:
Object
  • Object
show all
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.

bunny's wait_for_confirms, made to let go when the connection is lost.

A message written just as the connection dies is never confirmed, and bunny never tells the thread waiting for it so. Its recovery does try -- release_all_continuations wakes the waiting thread -- but the wait loops on an empty continuation queue and goes straight back to sleep, and the same recovery then replaces that queue with a new one, so no confirm can ever reach it. The thread sits out bunny's whole continuation timeout, fifteen seconds by default. Measured on bunny 2.24 and 3.4 alike, with bunny alone and through this library: a forced close at 86.45s, the connection back at 96.46s, the wait ended by Timeout::Error at 101.46s.

It costs more than the one message, because the wait holds the publish lock: every other publisher on the connection stalled behind it for those fifteen seconds, five of them after the connection was healthy again.

So Transport calls #abandon when a recovery attempt starts, and the waiting thread is told the connection was lost. The message stays a possible loss -- it was written and nothing answered for it -- so the publish fails as unconfirmed, not as declined.

The interruption is Thread#raise, which is only safe because it is fenced in: it can land only while the thread is inside wait_for_confirms, parked on a condition variable, and one that arrives as the wait is finishing is taken here and dropped rather than escaping later from somewhere else. The earliest bunny offers is before_recovery_attempt_starts, which comes one network_recovery_interval (5s) after the loss.

Defined Under Namespace

Classes: Abandoned

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ConfirmWait

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 a new instance of ConfirmWait.



247
248
249
250
 
# File 'lib/acemq/amqp/transport.rb', line 247

def initialize
  @lock = Mutex.new
  @waiter = nil
end

Instance Method Details

#abandon ⇒ 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.

Ends the wait in progress, if there is one.



268
269
270
271
272
273
274
275
 
# File 'lib/acemq/amqp/transport.rb', line 268

def abandon
  @lock.synchronize do
    @waiter&.raise(Abandoned,
                   "the connection was lost while waiting for the broker to confirm " \
                   "this message, so whether it arrived is unknown")
  end
  nil
end

#wait(channel) ⇒ 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.

bunny's answer to "were they all acked", unless #abandon came first.

Raises:

  • (Abandoned) —

    when the connection was lost during the wait



255
256
257
258
259
260
261
262
263
264
265
 
# File 'lib/acemq/amqp/transport.rb', line 255

def wait(channel)
  Thread.handle_interrupt(Abandoned => :never) do
    @lock.synchronize { @waiter = Thread.current }
    begin
      Thread.handle_interrupt(Abandoned => :immediate) { channel.wait_for_confirms }
    ensure
      @lock.synchronize { @waiter = nil }
      discard_late_abandon
    end
  end
end