diff options
Diffstat (limited to 'activesupport/lib/active_support/notifications/fanout.rb')
-rw-r--r-- | activesupport/lib/active_support/notifications/fanout.rb | 42 |
1 files changed, 27 insertions, 15 deletions
diff --git a/activesupport/lib/active_support/notifications/fanout.rb b/activesupport/lib/active_support/notifications/fanout.rb index 2e5bcf4639..6bf8c7d5de 100644 --- a/activesupport/lib/active_support/notifications/fanout.rb +++ b/activesupport/lib/active_support/notifications/fanout.rb @@ -1,4 +1,5 @@ require 'mutex_m' +require 'thread_safe' module ActiveSupport module Notifications @@ -11,7 +12,7 @@ module ActiveSupport def initialize @subscribers = [] - @listeners_for = {} + @listeners_for = ThreadSafe::Cache.new super end @@ -24,9 +25,15 @@ module ActiveSupport subscriber end - def unsubscribe(subscriber) + def unsubscribe(subscriber_or_name) synchronize do - @subscribers.reject! { |s| s.matches?(subscriber) } + case subscriber_or_name + when String + @subscribers.reject! { |s| s.matches?(subscriber_or_name) } + else + @subscribers.delete(subscriber_or_name) + end + @listeners_for.clear end end @@ -44,7 +51,9 @@ module ActiveSupport end def listeners_for(name) - synchronize do + # this is correctly done double-checked locking (ThreadSafe::Cache's lookups have volatile semantics) + @listeners_for[name] || synchronize do + # use synchronisation when accessing @subscribers @listeners_for[name] ||= @subscribers.select { |s| s.subscribed_to?(name) } end end @@ -76,6 +85,13 @@ module ActiveSupport def initialize(pattern, delegate) @pattern = pattern @delegate = delegate + @can_publish = delegate.respond_to?(:publish) + end + + def publish(name, *args) + if @can_publish + @delegate.publish name, *args + end end def start(name, id, payload) @@ -87,31 +103,27 @@ module ActiveSupport end def subscribed_to?(name) - @pattern === name.to_s + @pattern === name end - def matches?(subscriber_or_name) - self === subscriber_or_name || - @pattern && @pattern === subscriber_or_name + def matches?(name) + @pattern && @pattern === name end end class Timed < Evented - def initialize(pattern, delegate) - @timestack = [] - super - end - def publish(name, *args) @delegate.call name, *args end def start(name, id, payload) - @timestack.push Time.now + timestack = Thread.current[:_timestack] ||= [] + timestack.push Time.now end def finish(name, id, payload) - started = @timestack.pop + timestack = Thread.current[:_timestack] + started = timestack.pop @delegate.call(name, started, Time.now, id, payload) end end |