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.
Namespace
Methods
- F
- L
- N
- P
- S
- U
- W
Included Modules
- Mutex_m
Class Public methods
new() Link
Instance Public methods
finish(name, id, payload, listeners = listeners_for(name)) Link
listeners_for(name) Link
# 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) Link
publish(name, *args) Link
start(name, id, payload) Link
subscribe(pattern = nil, callable = nil, monotonic: false, &block) Link
# 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) Link
# 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