Class: ActiveSupport::Notifications::Fanout
Relationships & Source Files | |
Namespace Children | |
Modules:
| |
Classes:
| |
Super Chains via Extension / Inclusion / Inheritance | |
Instance Chain:
self,
FanoutIteration
|
|
Inherits: | Object |
Defined in: | activesupport/lib/active_support/notifications/fanout.rb |
Overview
This is a default queue implementation that ships with ::ActiveSupport::Notifications
. It just pushes events to all registered log subscribers.
This class is thread safe. All methods are reentrant.
Class Method Summary
- .new ⇒ Fanout constructor
Instance Method Summary
- #all_listeners_for(name)
- #build_handle(name, id, payload)
- #finish(name, id, payload, listeners = nil)
- #listeners_for(name)
- #listening?(name) ⇒ Boolean
- #publish(name)
- #publish_event(event)
- #start(name, id, payload)
- #subscribe(pattern = nil, callable = nil, monotonic: false, &block)
- #unsubscribe(subscriber_or_name)
-
#wait
This is a sync queue, so there is no waiting.
- #clear_cache(key = nil) Internal use only
- #group_listeners(listeners) Internal use only
- #groups_for(name) Internal use only
- #inspect Internal use only
FanoutIteration
- Included
Constructor Details
.new ⇒ Fanout
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 56
def initialize @mutex = Mutex.new @string_subscribers = Concurrent::Map.new { |h, k| h.compute_if_absent(k) { [] } } @other_subscribers = [] @all_listeners_for = Concurrent::Map.new @groups_for = Concurrent::Map.new end
Instance Method Details
#all_listeners_for(name)
[ GitHub ]# File 'activesupport/lib/active_support/notifications/fanout.rb', line 319
def all_listeners_for(name) # this is correctly done double-checked locking (Concurrent::Map's lookups have volatile semantics) @all_listeners_for[name] || @mutex.synchronize do # use synchronisation when accessing @subscribers @all_listeners_for[name] ||= @string_subscribers[name] + @other_subscribers.select { |s| s.subscribed_to?(name) } end end
#build_handle(name, id, payload)
[ GitHub ]# File 'activesupport/lib/active_support/notifications/fanout.rb', line 286
def build_handle(name, id, payload) groups = groups_for(name).map do |group_klass, grouped_listeners| group_klass.new(grouped_listeners, name, id, payload) end if groups.empty? NullHandle else Handle.new(self, name, id, groups, payload) end end
#clear_cache(key = nil)
This method is for internal use only.
[ GitHub ]
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 106
def clear_cache(key = nil) # :nodoc: if key @all_listeners_for.delete(key) @groups_for.delete(key) else @all_listeners_for.clear @groups_for.clear end end
#finish(name, id, payload, listeners = nil)
[ GitHub ]# File 'activesupport/lib/active_support/notifications/fanout.rb', line 305
def finish(name, id, payload, listeners = nil) handle_stack = IsolatedExecutionState[:_fanout_handle_stack] handle = handle_stack.pop handle.finish_with_values(name, id, payload) end
#group_listeners(listeners)
This method is for internal use only.
[ GitHub ]
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 190
def group_listeners(listeners) # :nodoc: listeners.group_by(&:group_class).transform_values do |s| s.map(&:delegate).freeze end.freeze end
#groups_for(name)
This method is for internal use only.
[ GitHub ]
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 196
def groups_for(name) # :nodoc: silenceable_groups, groups = @groups_for.compute_if_absent(name) do listeners = all_listeners_for(name) listeners.partition(&:silenceable).map { |l| group_listeners(l) } end unless silenceable_groups.empty? silenceable_groups.each do |group_class, subscriptions| active_subscriptions = subscriptions.reject { |s| s.silenced?(name) } unless active_subscriptions.empty? groups = groups.dup if groups.frozen? base_groups = groups[group_class] groups[group_class] = base_groups ? base_groups + active_subscriptions : active_subscriptions end end end groups end
#inspect
This method is for internal use only.
[ GitHub ]
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 64
def inspect # :nodoc: total_patterns = @string_subscribers.size + @other_subscribers.size "#<#{self.class} (#{total_patterns} patterns)>" end
#listeners_for(name)
[ GitHub ]# File 'activesupport/lib/active_support/notifications/fanout.rb', line 328
def listeners_for(name) all_listeners_for(name).reject { |s| s.silenced?(name) } end
#listening?(name) ⇒ Boolean
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 332
def listening?(name) all_listeners_for(name).any? { |s| !s.silenced?(name) } end
#publish(name)
[ GitHub ]# File 'activesupport/lib/active_support/notifications/fanout.rb', line 311
def publish(name, ...) iterate_guarding_exceptions(listeners_for(name)) { |s| s.publish(name, ...) } end
#publish_event(event)
[ GitHub ]# File 'activesupport/lib/active_support/notifications/fanout.rb', line 315
def publish_event(event) iterate_guarding_exceptions(listeners_for(event.name)) { |s| s.publish_event(event) } end
#start(name, id, payload)
[ GitHub ]# File 'activesupport/lib/active_support/notifications/fanout.rb', line 298
def start(name, id, payload) handle_stack = (IsolatedExecutionState[:_fanout_handle_stack] ||= []) handle = build_handle(name, id, payload) handle_stack << handle handle.start end
#subscribe(pattern = nil, callable = nil, monotonic: false, &block)
[ GitHub ]# File 'activesupport/lib/active_support/notifications/fanout.rb', line 69
def subscribe(pattern = nil, callable = nil, monotonic: false, &block) subscriber = Subscribers.new(pattern, callable || block, monotonic) @mutex.synchronize do case pattern when String @string_subscribers[pattern] << subscriber clear_cache(pattern) when NilClass, Regexp @other_subscribers << subscriber clear_cache else raise ArgumentError, "pattern must be specified as a String, Regexp or empty" end end subscriber end
#unsubscribe(subscriber_or_name)
[ GitHub ]# File 'activesupport/lib/active_support/notifications/fanout.rb', line 86
def unsubscribe(subscriber_or_name) @mutex.synchronize do case subscriber_or_name when String @string_subscribers[subscriber_or_name].clear clear_cache(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) clear_cache(pattern) else @other_subscribers.delete(subscriber_or_name) clear_cache end end end end
#wait
This is a sync queue, so there is no waiting.
# File 'activesupport/lib/active_support/notifications/fanout.rb', line 337
def wait end