Class: Qsagi::ConfirmedQueue
- Inherits:
-
Object
- Object
- Qsagi::ConfirmedQueue
- Defined in:
- lib/qsagi/confirmed_queue.rb
Instance Attribute Summary collapse
-
#nacked_messages ⇒ Object
readonly
Returns the value of attribute nacked_messages.
Instance Method Summary collapse
- #_channel ⇒ Object
- #_confirm_messages!(attributes) ⇒ Object
- #_confirm_select ⇒ Object
- #_wait_for_confirms? ⇒ Boolean
- #connect ⇒ Object
- #disconnect ⇒ Object
-
#initialize(queue) ⇒ ConfirmedQueue
constructor
A new instance of ConfirmedQueue.
- #pop(opts = {}) ⇒ Object
- #push(message) ⇒ Object
- #wait_for_confirms ⇒ Object
Constructor Details
#initialize(queue) ⇒ ConfirmedQueue
Returns a new instance of ConfirmedQueue.
7 8 9 10 11 12 13 |
# File 'lib/qsagi/confirmed_queue.rb', line 7 def initialize(queue) @queue = queue @nacked_messages = [] @unconfirmed_messages = {} @wait_for_confirms = false @semaphore = Mutex.new end |
Instance Attribute Details
#nacked_messages ⇒ Object (readonly)
Returns the value of attribute nacked_messages.
5 6 7 |
# File 'lib/qsagi/confirmed_queue.rb', line 5 def @nacked_messages end |
Instance Method Details
#_channel ⇒ Object
41 42 43 |
# File 'lib/qsagi/confirmed_queue.rb', line 41 def _channel @queue.channel end |
#_confirm_messages!(attributes) ⇒ Object
45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 |
# File 'lib/qsagi/confirmed_queue.rb', line 45 def (attributes) if attributes[:is_nack] if attributes[:multiple] @nacked_messages += @unconfirmed_messages.select { |k,v| k <= attributes[:delivery_tag] }.values @unconfirmed_messages.delete_if { |k,v| k <= attributes[:delivery_tag] } else @nacked_messages << @unconfirmed_messages.delete(attributes[:delivery_tag]) end else if attributes[:multiple] @unconfirmed_messages.delete_if { |k,v| k <= attributes[:delivery_tag] } else @unconfirmed_messages.delete(attributes[:delivery_tag]) end end end |
#_confirm_select ⇒ Object
62 63 64 65 66 67 68 69 70 |
# File 'lib/qsagi/confirmed_queue.rb', line 62 def _confirm_select callback = lambda do |delivery_tag, multiple, is_nack| @semaphore.synchronize do (:delivery_tag => delivery_tag, :multiple => multiple, :is_nack => is_nack) end end _channel.confirm_select(callback) end |
#_wait_for_confirms? ⇒ Boolean
72 73 74 |
# File 'lib/qsagi/confirmed_queue.rb', line 72 def _wait_for_confirms? @wait_for_confirms end |
#connect ⇒ Object
15 16 17 18 |
# File 'lib/qsagi/confirmed_queue.rb', line 15 def connect @queue.connect _confirm_select end |
#disconnect ⇒ Object
20 21 22 |
# File 'lib/qsagi/confirmed_queue.rb', line 20 def disconnect @queue.disconnect end |
#pop(opts = {}) ⇒ Object
33 34 35 |
# File 'lib/qsagi/confirmed_queue.rb', line 33 def pop(opts={}) @queue.pop(opts) end |
#push(message) ⇒ Object
24 25 26 27 28 29 30 31 |
# File 'lib/qsagi/confirmed_queue.rb', line 24 def push() next_sequence_number = _channel.next_publish_seq_no @semaphore.synchronize do @unconfirmed_messages[next_sequence_number] = end @queue.push() @wait_for_confirms = true end |
#wait_for_confirms ⇒ Object
37 38 39 |
# File 'lib/qsagi/confirmed_queue.rb', line 37 def wait_for_confirms _channel.wait_for_confirms if _wait_for_confirms? end |