class ActiveSupport::Notifications::Fanout
This class is thread safe. All methods are reentrant.
It just pushes events to all registered log subscribers.
This is a default queue implementation that ships with Notifications.
def finish(name, id, payload, listeners = listeners_for(name))
def finish(name, id, payload, listeners = listeners_for(name)) iterate_guarding_exceptions(listeners) { |s| s.finish(name, id, payload) } end
def initialize
def initialize @string_subscribers = Hash.new { |h, k| h[k] = [] } @other_subscribers = [] @listeners_for = Concurrent::Map.new super end
def iterate_guarding_exceptions(listeners)
def iterate_guarding_exceptions(listeners) exceptions = nil listeners.each do |s| yield s rescue Exception => e exceptions ||= [] exceptions << e end if exceptions if exceptions.size == 1 raise exceptions.first else raise InstrumentationSubscriberError.new(exceptions), cause: exceptions.first end end listeners end
def listeners_for(name)
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
def listening?(name)
def listening?(name) listeners_for(name).any? end
def publish(name, *args)
def publish(name, *args) iterate_guarding_exceptions(listeners_for(name)) { |s| s.publish(name, *args) } end
def publish_event(event)
def publish_event(event) iterate_guarding_exceptions(listeners_for(event.name)) { |s| s.publish_event(event) } end
def start(name, id, payload)
def start(name, id, payload) iterate_guarding_exceptions(listeners_for(name)) { |s| s.start(name, id, payload) } end
def subscribe(pattern = nil, callable = nil, monotonic: false, &block)
def subscribe(pattern = nil, callable = nil, monotonic: false, &block) subscriber = Subscribers.new(pattern, callable || block, monotonic) synchronize do case pattern when String @string_subscribers[pattern] << subscriber @listeners_for.delete(pattern) when NilClass, Regexp @other_subscribers << subscriber @listeners_for.clear else raise ArgumentError, "pattern must be specified as a String, Regexp or empty" end end subscriber end
def unsubscribe(subscriber_or_name)
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
def wait
def wait end