Class: AceMQ::AMQP::ReturnedMessages 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.

The messages the broker handed back, on their way to whoever sent them.

A class of its own because it is the one place in this file where two threads meet: bunny dispatches basic.return on its reader thread while the thread that published is still inside Transport#publish waiting for a confirm. It is also the only place that has to know bunny's rule for delivering a return, which is a rule about exchange objects rather than about messages.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ReturnedMessages

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 ReturnedMessages.



428
429
430
431
 
# File 'lib/acemq/amqp/transport.rb', line 428

def initialize
  @returns = Thread::Queue.new
  @watched = []
end

Instance Method Details

#arm(channel, exchange) ⇒ 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.

Gets ready to hear about this exchange, and forgets anything left over.

bunny hands basic.return to the Bunny::Exchange object registered on the channel under that name and drops it with a warning when there is none, so there has to be one — and it has to be one that does not declare, because this transport publishes to exchanges the application declared and re-declaring one with a type guessed here is how a publish turns into PRECONDITION_FAILED. no_declare is bunny's word for that; the default exchange skips the declaration on its own name anyway.

Registered once per exchange. The object is kept by the channel, which is what makes throwing this one away safe.

Anything still queued belongs to a publish that raised before it could read its own return, and attributing it to this message would report the wrong one as unroutable.



449
450
451
452
 
# File 'lib/acemq/amqp/transport.rb', line 449

def arm(channel, exchange)
  watch(channel, exchange)
  @returns.clear
end

#arm_all(channel, 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.

The same, for a batch: every exchange it publishes to is watched before any of it goes out, and whatever was left over is dropped once rather than between two publishes. A return for the message just sent arrives while the next one is still being written, and clearing per message would throw it away.



459
460
461
462
 
# File 'lib/acemq/amqp/transport.rb', line 459

def arm_all(channel, exchanges)
  exchanges.each { |exchange| watch(channel, exchange) }
  @returns.clear
end

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

Forgets which exchanges are watched, for a channel that has been reopened: the registrations lived on the old one and went with it.



466
467
468
469
 
# File 'lib/acemq/amqp/transport.rb', line 466

def reopened
  @watched = []
  @returns.clear
end

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

Why the message just published was handed back, or nil when it was not.

Non-blocking, because a return that was coming has already arrived: the broker sends it ahead of the confirm and bunny dispatches frames in order, so a wait here would put the cost of mandatory on every message rather than on the ones it is about.



477
478
479
480
481
 
# File 'lib/acemq/amqp/transport.rb', line 477

def take
  @returns.pop(true)&.last
rescue ThreadError
  nil
end

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

Everything queued, as pairs of message id and reason.

A batch has to tell which of its messages came back rather than only that one did, and the only thing that can say so is the returned message's own message-id — the returns arrive in whatever order the broker sends them, and nothing about position survives the trip.



489
490
491
492
493
494
 
# File 'lib/acemq/amqp/transport.rb', line 489

def take_all
  taken = []
  loop { taken << @returns.pop(true) }
rescue ThreadError
  taken
end