Class: Vx::Consumer::Subscriber
- Inherits:
-
Bunny::Consumer
- Object
- Bunny::Consumer
- Vx::Consumer::Subscriber
- Includes:
- Instrument
- Defined in:
- lib/vx/consumer/subscriber.rb
Instance Attribute Summary collapse
-
#queue_name ⇒ Object
Returns the value of attribute queue_name.
-
#vx_consumer_name ⇒ Object
Returns the value of attribute vx_consumer_name.
Instance Method Summary collapse
- #call(*args) ⇒ Object
- #cancel ⇒ Object
- #closed? ⇒ Boolean
- #graceful_shutdown ⇒ Object
- #in_progress ⇒ Object
-
#initialize(*args) ⇒ Subscriber
constructor
A new instance of Subscriber.
- #join ⇒ Object
- #running? ⇒ Boolean
- #try_graceful_shutdown ⇒ Object
- #wait_shutdown ⇒ Object
Methods included from Instrument
Constructor Details
#initialize(*args) ⇒ Subscriber
Returns a new instance of Subscriber.
12 13 14 15 |
# File 'lib/vx/consumer/subscriber.rb', line 12 def initialize(*args) super(*args) @lock = Mutex.new end |
Instance Attribute Details
#queue_name ⇒ Object
Returns the value of attribute queue_name.
10 11 12 |
# File 'lib/vx/consumer/subscriber.rb', line 10 def queue_name @queue_name end |
#vx_consumer_name ⇒ Object
Returns the value of attribute vx_consumer_name.
10 11 12 |
# File 'lib/vx/consumer/subscriber.rb', line 10 def vx_consumer_name @vx_consumer_name end |
Instance Method Details
#call(*args) ⇒ Object
49 50 51 52 53 54 |
# File 'lib/vx/consumer/subscriber.rb', line 49 def call(*args) in_progress do @on_delivery.call(*args) if @on_delivery sleep 0 end end |
#cancel ⇒ Object
56 57 58 59 60 61 62 |
# File 'lib/vx/consumer/subscriber.rb', line 56 def cancel instrument('cancel_consumer', consumer: vx_consumer_name, channel: channel.id) unless closed? super channel.close unless closed? end end |
#closed? ⇒ Boolean
64 65 66 |
# File 'lib/vx/consumer/subscriber.rb', line 64 def closed? channel.closed? end |
#graceful_shutdown ⇒ Object
17 18 19 20 21 22 23 |
# File 'lib/vx/consumer/subscriber.rb', line 17 def graceful_shutdown instrument('try_graceful_shutdown_consumer', consumer: vx_consumer_name) in_progress do cancel instrument('graceful_shutdown_consumer', consumer: vx_consumer_name) end end |
#in_progress ⇒ Object
39 40 41 42 43 |
# File 'lib/vx/consumer/subscriber.rb', line 39 def in_progress @lock.synchronize do yield end end |
#join ⇒ Object
68 69 70 |
# File 'lib/vx/consumer/subscriber.rb', line 68 def join channel.work_pool.join end |
#running? ⇒ Boolean
45 46 47 |
# File 'lib/vx/consumer/subscriber.rb', line 45 def running? @lock.locked? end |
#try_graceful_shutdown ⇒ Object
25 26 27 28 29 30 31 32 33 34 35 36 37 |
# File 'lib/vx/consumer/subscriber.rb', line 25 def try_graceful_shutdown if @lock.try_lock begin instrument('graceful_shutdown_consumer', consumer: vx_consumer_name) cancel ensure @lock.unlock end true else false end end |
#wait_shutdown ⇒ Object
72 73 74 75 76 77 78 |
# File 'lib/vx/consumer/subscriber.rb', line 72 def wait_shutdown Thread.new do Thread.current.abort_on_exception = true Consumer.wait_shutdown cancel end end |