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