2009-09-16 22:53:49 +00:00
|
|
|
require 'thread'
|
2009-10-01 22:00:22 +00:00
|
|
|
require 'active_support/core_ext/module/delegation'
|
2009-09-16 22:53:49 +00:00
|
|
|
|
|
|
|
module ActiveSupport
|
2009-10-15 21:06:15 +00:00
|
|
|
# Orchestra provides an instrumentation API for Ruby. To instrument an action
|
2009-10-06 12:42:42 +00:00
|
|
|
# in Ruby you just need to do:
|
2009-09-16 22:53:49 +00:00
|
|
|
#
|
2009-10-15 21:06:15 +00:00
|
|
|
# ActiveSupport::Orchestra.instrument(:render, :extra => :information) do
|
2009-09-16 22:53:49 +00:00
|
|
|
# render :text => "Foo"
|
|
|
|
# end
|
|
|
|
#
|
2009-10-06 12:42:42 +00:00
|
|
|
# You can consume those events and the information they provide by registering
|
|
|
|
# a subscriber. For instance, let's store all instrumented events in an array:
|
2009-09-16 22:53:49 +00:00
|
|
|
#
|
2009-10-06 12:42:42 +00:00
|
|
|
# @events = []
|
|
|
|
#
|
|
|
|
# ActiveSupport::Orchestra.subscribe do |event|
|
|
|
|
# @events << event
|
|
|
|
# end
|
2009-09-16 22:53:49 +00:00
|
|
|
#
|
2009-10-15 21:06:15 +00:00
|
|
|
# ActiveSupport::Orchestra.instrument(:render, :extra => :information) do
|
2009-09-16 22:53:49 +00:00
|
|
|
# render :text => "Foo"
|
|
|
|
# end
|
|
|
|
#
|
2009-10-06 12:42:42 +00:00
|
|
|
# event = @events.first
|
|
|
|
# event.class #=> ActiveSupport::Orchestra::Event
|
2009-09-16 22:53:49 +00:00
|
|
|
# event.name #=> :render
|
|
|
|
# event.duration #=> 10 (in miliseconds)
|
|
|
|
# event.result #=> "Foo"
|
|
|
|
# event.payload #=> { :extra => :information }
|
|
|
|
#
|
2009-10-06 12:42:42 +00:00
|
|
|
# When subscribing to Orchestra, you can pass a pattern, to only consume
|
|
|
|
# events that match the pattern:
|
|
|
|
#
|
|
|
|
# ActiveSupport::Orchestra.subscribe(/render/) do |event|
|
|
|
|
# @render_events << event
|
|
|
|
# end
|
|
|
|
#
|
|
|
|
# Orchestra ships with a queue implementation that consumes and publish events
|
|
|
|
# to subscribers in a thread. You can use any queue implementation you want.
|
2009-09-16 22:53:49 +00:00
|
|
|
#
|
2009-10-15 21:06:15 +00:00
|
|
|
module Orchestra
|
2009-09-30 11:59:15 +00:00
|
|
|
mattr_accessor :queue
|
|
|
|
|
2009-10-01 22:00:22 +00:00
|
|
|
class << self
|
|
|
|
delegate :instrument, :to => :instrumenter
|
|
|
|
|
|
|
|
def instrumenter
|
|
|
|
Thread.current[:orchestra_instrumeter] ||= Instrumenter.new(publisher)
|
|
|
|
end
|
|
|
|
|
|
|
|
def publisher
|
|
|
|
@publisher ||= Publisher.new(queue)
|
|
|
|
end
|
|
|
|
|
|
|
|
def subscribe(pattern=nil, &block)
|
|
|
|
Subscriber.new(queue).bind(pattern).subscribe(&block)
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
class Instrumenter
|
|
|
|
def initialize(publisher)
|
|
|
|
@publisher = publisher
|
|
|
|
end
|
|
|
|
|
2009-10-02 00:14:38 +00:00
|
|
|
def instrument(name, payload={})
|
|
|
|
payload[:time] = Time.now
|
|
|
|
payload[:thread_id] = Thread.current.object_id
|
2009-10-07 14:17:50 +00:00
|
|
|
payload[:result] = yield if block_given?
|
2009-10-01 22:00:22 +00:00
|
|
|
ensure
|
2009-10-02 00:14:38 +00:00
|
|
|
payload[:duration] = 1000 * (Time.now.to_f - payload[:time].to_f)
|
|
|
|
@publisher.publish(name, payload)
|
2009-10-01 22:00:22 +00:00
|
|
|
end
|
2009-09-16 22:53:49 +00:00
|
|
|
end
|
|
|
|
|
2009-10-01 22:00:22 +00:00
|
|
|
class Publisher
|
|
|
|
def initialize(queue)
|
|
|
|
@queue = queue
|
|
|
|
end
|
|
|
|
|
2009-10-02 00:14:38 +00:00
|
|
|
def publish(name, payload)
|
|
|
|
@queue.publish(name, payload)
|
2009-10-01 22:00:22 +00:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
class Subscriber
|
|
|
|
def initialize(queue)
|
|
|
|
@queue = queue
|
|
|
|
end
|
|
|
|
|
|
|
|
def bind(pattern)
|
|
|
|
@pattern = pattern
|
|
|
|
self
|
|
|
|
end
|
|
|
|
|
|
|
|
def subscribe
|
2009-10-02 00:14:38 +00:00
|
|
|
@queue.subscribe(@pattern) do |name, payload|
|
|
|
|
yield Event.new(name, payload)
|
2009-10-01 22:00:22 +00:00
|
|
|
end
|
|
|
|
end
|
2009-09-16 22:53:49 +00:00
|
|
|
end
|
|
|
|
|
|
|
|
class Event
|
2009-10-02 00:14:38 +00:00
|
|
|
attr_reader :name, :time, :duration, :thread_id, :result, :payload
|
2009-09-16 22:53:49 +00:00
|
|
|
|
2009-10-02 00:14:38 +00:00
|
|
|
def initialize(name, payload)
|
2009-09-16 22:53:49 +00:00
|
|
|
@name = name
|
2009-10-02 00:14:38 +00:00
|
|
|
@payload = payload.dup
|
|
|
|
@time = @payload.delete(:time)
|
|
|
|
@thread_id = @payload.delete(:thread_id)
|
|
|
|
@result = @payload.delete(:result)
|
|
|
|
@duration = @payload.delete(:duration)
|
2009-09-16 22:53:49 +00:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2009-09-30 11:59:15 +00:00
|
|
|
# This is a default queue implementation that ships with Orchestra. It
|
|
|
|
# consumes events in a thread and publish them to all registered subscribers.
|
|
|
|
#
|
|
|
|
class LittleFanout
|
2009-09-16 22:53:49 +00:00
|
|
|
def initialize
|
2009-10-06 12:42:42 +00:00
|
|
|
@listeners, @stream = [], Queue.new
|
|
|
|
@thread = Thread.new { consume }
|
2009-09-16 22:53:49 +00:00
|
|
|
end
|
|
|
|
|
2009-10-02 00:14:38 +00:00
|
|
|
def publish(*event)
|
2009-09-30 11:59:15 +00:00
|
|
|
@stream.push(event)
|
2009-09-16 22:53:49 +00:00
|
|
|
end
|
|
|
|
|
2009-09-30 11:59:15 +00:00
|
|
|
def subscribe(pattern=nil, &block)
|
|
|
|
@listeners << Listener.new(pattern, &block)
|
2009-09-16 22:53:49 +00:00
|
|
|
end
|
|
|
|
|
2009-10-06 12:42:42 +00:00
|
|
|
def consume
|
|
|
|
while event = @stream.shift
|
|
|
|
@listeners.each { |l| l.publish(*event) }
|
|
|
|
end
|
2009-09-30 11:59:15 +00:00
|
|
|
end
|
|
|
|
|
|
|
|
class Listener
|
2009-10-06 12:42:42 +00:00
|
|
|
attr_reader :thread
|
|
|
|
|
2009-09-30 11:59:15 +00:00
|
|
|
def initialize(pattern, &block)
|
|
|
|
@pattern = pattern
|
|
|
|
@subscriber = block
|
2009-10-06 12:42:42 +00:00
|
|
|
@queue = Queue.new
|
|
|
|
@thread = Thread.new { consume }
|
2009-09-30 11:59:15 +00:00
|
|
|
end
|
|
|
|
|
2009-10-02 00:14:38 +00:00
|
|
|
def publish(name, payload)
|
2009-10-07 14:17:50 +00:00
|
|
|
unless @pattern && !(@pattern === name.to_s)
|
2009-10-06 12:42:42 +00:00
|
|
|
@queue << [name, payload]
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
def consume
|
|
|
|
while event = @queue.shift
|
|
|
|
@subscriber.call(*event)
|
2009-09-30 11:59:15 +00:00
|
|
|
end
|
|
|
|
end
|
2009-09-16 22:53:49 +00:00
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
2009-09-30 11:59:15 +00:00
|
|
|
|
|
|
|
Orchestra.queue = Orchestra::LittleFanout.new
|
2009-09-16 22:53:49 +00:00
|
|
|
end
|