aboutsummaryrefslogtreecommitdiffstats
path: root/activesupport/lib/active_support/notifications/fanout.rb
diff options
context:
space:
mode:
Diffstat (limited to 'activesupport/lib/active_support/notifications/fanout.rb')
-rw-r--r--activesupport/lib/active_support/notifications/fanout.rb36
1 files changed, 3 insertions, 33 deletions
diff --git a/activesupport/lib/active_support/notifications/fanout.rb b/activesupport/lib/active_support/notifications/fanout.rb
index bb07e4765c..ac482a2ec8 100644
--- a/activesupport/lib/active_support/notifications/fanout.rb
+++ b/activesupport/lib/active_support/notifications/fanout.rb
@@ -3,11 +3,9 @@ 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.
- #
+ # just pushes events to all registered subscribers.
class Fanout
- def initialize(sync = false)
- @subscriber_klass = sync ? Subscriber : AsyncSubscriber
+ def initialize
@subscribers = []
end
@@ -16,7 +14,7 @@ module ActiveSupport
end
def subscribe(pattern = nil, &block)
- @subscribers << @subscriber_klass.new(pattern, &block)
+ @subscribers << Subscriber.new(pattern, &block)
end
def publish(*args)
@@ -68,34 +66,6 @@ module ActiveSupport
@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