class AS::Notifications::Fanout

This is a default queue implementation that ships with Notifications. It just pushes events to all registered log subscribers.

This class is thread safe. All methods are reentrant.

Public Class Methods

new() click to toggle source
Calls superclass method
# File lib/as/notifications/fanout.rb, line 12
def initialize
  @subscribers = []
  @listeners_for = {}
  super
end

Public Instance Methods

finish(name, id, payload) click to toggle source
# File lib/as/notifications/fanout.rb, line 38
def finish(name, id, payload)
  listeners_for(name).each { |s| s.finish(name, id, payload) }
end
listeners_for(name) click to toggle source
# File lib/as/notifications/fanout.rb, line 46
def listeners_for(name)
  synchronize do
    @listeners_for[name] ||= @subscribers.select { |s| s.subscribed_to?(name) }
  end
end
listening?(name) click to toggle source
# File lib/as/notifications/fanout.rb, line 52
def listening?(name)
  listeners_for(name).any?
end
publish(name, *args) click to toggle source
# File lib/as/notifications/fanout.rb, line 42
def publish(name, *args)
  listeners_for(name).each { |s| s.publish(name, *args) }
end
start(name, id, payload) click to toggle source
# File lib/as/notifications/fanout.rb, line 34
def start(name, id, payload)
  listeners_for(name).each { |s| s.start(name, id, payload) }
end
subscribe(pattern = nil, callable = nil, &block) click to toggle source
# File lib/as/notifications/fanout.rb, line 18
def subscribe(pattern = nil, callable = nil, &block)
  subscriber = Subscribers.new pattern, callable || block
  synchronize do
    @subscribers << subscriber
    @listeners_for.clear
  end
  subscriber
end
unsubscribe(subscriber) click to toggle source
# File lib/as/notifications/fanout.rb, line 27
def unsubscribe(subscriber)
  synchronize do
    @subscribers.reject! { |s| s.matches?(subscriber) }
    @listeners_for.clear
  end
end
wait() click to toggle source

This is a sync queue, so there is no waiting.

# File lib/as/notifications/fanout.rb, line 57
def wait
end