Class: ActiveSupport::Notifications::Fanout
- Includes:
- Mutex_m
- Defined in:
- activesupport/lib/active_support/notifications/fanout.rb
Overview
This is a default queue implementation that ships with Notifications. It just pushes events to all registered log subscribers.
This class is thread safe. All methods are reentrant.
Defined Under Namespace
Modules: Subscribers
Instance Method Summary collapse
- #finish(name, id, payload, listeners = listeners_for(name)) ⇒ Object
-
#initialize ⇒ Fanout
constructor
A new instance of Fanout.
- #listeners_for(name) ⇒ Object
- #listening?(name) ⇒ Boolean
- #publish(name, *args) ⇒ Object
- #start(name, id, payload) ⇒ Object
- #subscribe(pattern = nil, callable = nil, monotonic: false, &block) ⇒ Object
- #unsubscribe(subscriber_or_name) ⇒ Object
-
#wait ⇒ Object
This is a sync queue, so there is no waiting.
Constructor Details
Instance Method Details
#finish(name, id, payload, listeners = listeners_for(name)) ⇒ Object
62 63 64 |
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 62 def finish(name, id, payload, listeners = listeners_for(name)) listeners.each { |s| s.finish(name, id, payload) } end |
#listeners_for(name) ⇒ Object
70 71 72 73 74 75 76 77 |
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 70 def listeners_for(name) # this is correctly done double-checked locking (Concurrent::Map's lookups have volatile semantics) @listeners_for[name] || synchronize do # use synchronisation when accessing @subscribers @listeners_for[name] ||= @string_subscribers[name] + @other_subscribers.select { |s| s.subscribed_to?(name) } end end |
#listening?(name) ⇒ Boolean
79 80 81 |
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 79 def listening?(name) listeners_for(name).any? end |
#publish(name, *args) ⇒ Object
66 67 68 |
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 66 def publish(name, *args) listeners_for(name).each { |s| s.publish(name, *args) } end |
#start(name, id, payload) ⇒ Object
58 59 60 |
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 58 def start(name, id, payload) listeners_for(name).each { |s| s.start(name, id, payload) } end |
#subscribe(pattern = nil, callable = nil, monotonic: false, &block) ⇒ Object
24 25 26 27 28 29 30 31 32 33 34 35 36 |
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 24 def subscribe(pattern = nil, callable = nil, monotonic: false, &block) subscriber = Subscribers.new(pattern, callable || block, monotonic) synchronize do if String === pattern @string_subscribers[pattern] << subscriber @listeners_for.delete(pattern) else @other_subscribers << subscriber @listeners_for.clear end end subscriber end |
#unsubscribe(subscriber_or_name) ⇒ Object
38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 |
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 38 def unsubscribe(subscriber_or_name) synchronize do case subscriber_or_name when String @string_subscribers[subscriber_or_name].clear @listeners_for.delete(subscriber_or_name) @other_subscribers.each { |sub| sub.unsubscribe!(subscriber_or_name) } else pattern = subscriber_or_name.try(:pattern) if String === pattern @string_subscribers[pattern].delete(subscriber_or_name) @listeners_for.delete(pattern) else @other_subscribers.delete(subscriber_or_name) @listeners_for.clear end end end end |
#wait ⇒ Object
This is a sync queue, so there is no waiting.
84 85 |
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 84 def wait end |