Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Merge branch 'remote'

* remote:
  move fanout back to a global variable, add a mutex for safety
  • Loading branch information...
commit 056dbf4f114eb20094c578793b4b4ae312e43774 2 parents 7896f35 + bf8e205
@tenderlove tenderlove authored
View
31 activesupport/lib/active_support/notifications.rb
@@ -135,24 +135,9 @@ module ActiveSupport
# to log subscribers in a thread. You can use any queue implementation you want.
#
module Notifications
- class Registry # :nodoc:
- def self.instance
- Thread.current[name] ||= new
- end
-
- attr_reader :notifier, :instrumenter
-
- def initialize
- self.notifier = Fanout.new
- end
-
- def notifier=(notifier)
- @notifier = notifier
- @instrumenter = Instrumenter.new(notifier)
- end
- end
-
class << self
+ attr_accessor :notifier
+
def publish(name, *args)
notifier.publish(name, *args)
end
@@ -181,16 +166,10 @@ def unsubscribe(args)
end
def instrumenter
- Registry.instance.instrumenter
- end
-
- def notifier
- Registry.instance.notifier
- end
-
- def notifier=(notifier)
- Registry.instance.notifier = notifier
+ Thread.current[:"instrumentation_#{notifier.object_id}"] ||= Instrumenter.new(notifier)
end
end
+
+ self.notifier = Fanout.new
end
end
View
24 activesupport/lib/active_support/notifications/fanout.rb
@@ -1,26 +1,34 @@
+require 'mutex_m'
+
module ActiveSupport
module Notifications
# This is a default queue implementation that ships with Notifications.
# It just pushes events to all registered log subscribers.
#
- # Only one of these objects should instantiated per thread. Concurrent
- # access to this class is not allowed.
+ # This class is thread safe. All methods are reentrant.
class Fanout
+ include Mutex_m
+
def initialize
@subscribers = []
@listeners_for = {}
+ super
end
def subscribe(pattern = nil, block = Proc.new)
subscriber = Subscribers.new pattern, block
- @subscribers << subscriber
- @listeners_for.clear
+ synchronize do
+ @subscribers << subscriber
+ @listeners_for.clear
+ end
subscriber
end
def unsubscribe(subscriber)
- @subscribers.reject! { |s| s.matches?(subscriber) }
- @listeners_for.clear
+ synchronize do
+ @subscribers.reject! { |s| s.matches?(subscriber) }
+ @listeners_for.clear
+ end
end
def start(name, id, payload)
@@ -36,7 +44,9 @@ def publish(name, *args)
end
def listeners_for(name)
- @listeners_for[name] ||= @subscribers.select { |s| s.subscribed_to?(name) }
+ synchronize do
+ @listeners_for[name] ||= @subscribers.select { |s| s.subscribed_to?(name) }
+ end
end
def listening?(name)
Please sign in to comment.
Something went wrong with that request. Please try again.