diff options
author | Pratik Naik <pratiknaik@gmail.com> | 2010-01-04 03:24:39 +0530 |
---|---|---|
committer | Pratik Naik <pratiknaik@gmail.com> | 2010-01-04 03:24:39 +0530 |
commit | cda36a0731f14b33a920bf7e32255661e06f890a (patch) | |
tree | 79ccba37953f9fe3055503be42b1610faa6d64ad /activesupport/lib/active_support/notifications/fanout.rb | |
parent | bd4a3cce4ecd8e648179a91e26506e3622ac2162 (diff) | |
parent | a115b5d79a850bb56cd3c9db9a05d6da35e3d7be (diff) | |
download | rails-cda36a0731f14b33a920bf7e32255661e06f890a.tar.gz rails-cda36a0731f14b33a920bf7e32255661e06f890a.tar.bz2 rails-cda36a0731f14b33a920bf7e32255661e06f890a.zip |
Merge remote branch 'mainstream/master'
Diffstat (limited to 'activesupport/lib/active_support/notifications/fanout.rb')
-rw-r--r-- | activesupport/lib/active_support/notifications/fanout.rb | 101 |
1 files changed, 101 insertions, 0 deletions
diff --git a/activesupport/lib/active_support/notifications/fanout.rb b/activesupport/lib/active_support/notifications/fanout.rb new file mode 100644 index 0000000000..bb07e4765c --- /dev/null +++ b/activesupport/lib/active_support/notifications/fanout.rb @@ -0,0 +1,101 @@ +require 'thread' + +module ActiveSupport + module Notifications + # This is a default queue implementation that ships with Notifications. It + # consumes events in a thread and publish them to all registered subscribers. + # + class Fanout + def initialize(sync = false) + @subscriber_klass = sync ? Subscriber : AsyncSubscriber + @subscribers = [] + end + + def bind(pattern) + Binding.new(self, pattern) + end + + def subscribe(pattern = nil, &block) + @subscribers << @subscriber_klass.new(pattern, &block) + end + + def publish(*args) + @subscribers.each { |s| s.publish(*args) } + end + + def wait + sleep(0.05) until @subscribers.all?(&:drained?) + end + + # Used for internal implementation only. + class Binding #:nodoc: + def initialize(queue, pattern) + @queue = queue + @pattern = + case pattern + when Regexp, NilClass + pattern + else + /^#{Regexp.escape(pattern.to_s)}/ + end + end + + def subscribe(&block) + @queue.subscribe(@pattern, &block) + end + end + + class Subscriber #:nodoc: + def initialize(pattern, &block) + @pattern = pattern + @block = block + end + + def publish(*args) + push(*args) if matches?(args.first) + end + + def drained? + true + end + + private + def matches?(name) + !@pattern || @pattern =~ name.to_s + end + + def push(*args) + @block.call(*args) + end + end + + # Used for internal implementation only. + class AsyncSubscriber < Subscriber #:nodoc: + def initialize(pattern, &block) + super + @events = Queue.new + start_consumer + end + + def drained? + @events.empty? + end + + private + def start_consumer + Thread.new { consume } + end + + def consume + while args = @events.shift + @block.call(*args) + end + end + + def push(*args) + @events << args + end + end + end + end +end |