Added Orchestra.

activesupport/lib/active_support/autoload.rb

@@ -17,6 +17,7 @@ module ActiveSupport
   autoload :Multibyte, 'active_support/multibyte'
   autoload :NewCallbacks, 'active_support/new_callbacks'
   autoload :OptionMerger, 'active_support/option_merger'
+  autoload :Orchestra, 'active_support/orchestra'
   autoload :OrderedHash, 'active_support/ordered_hash'
   autoload :OrderedOptions, 'active_support/ordered_options'
   autoload :Rescuable, 'active_support/rescuable'

activesupport/lib/active_support/orchestra.rb

+require 'thread'
+
+module ActiveSupport
+  # Orchestra provides an instrumentation API for Ruby. To instrument an action
+  # in Ruby you just need to:
+  #
+  #   ActiveSupport::Orchestra.instrument(:render, :extra => :information) do
+  #     render :text => "Foo"
+  #   end
+  #
+  # Those actions are consumed by listeners. A listener is anything that responds
+  # to push. You can even register an array:
+  #
+  #   @listener = []
+  #   ActiveSupport::Orchestra.register @listener
+  #
+  #   ActiveSupport::Orchestra.instrument(:render, :extra => :information) do
+  #     render :text => "Foo"
+  #   end
+  #
+  #   event           #=> ActiveSupport::Orchestra::Event
+  #   event.name      #=> :render
+  #   event.duration  #=> 10 (in miliseconds)
+  #   event.result    #=> "Foo"
+  #   event.payload   #=> { :extra => :information }
+  #
+  # Orchestra ships with a default listener implementation which puts events in
+  # a stream and consume them in a Thread. This implementation is thread safe
+  # and is available at ActiveSupport::Orchestra::Listener.
+  #
+  module Orchestra
+    @stacked_events = Hash.new { |h,k| h[k] = [] }
+    @listeners = []
+
+    def self.instrument(name, payload=nil)
+      stack = @stacked_events[Thread.current.object_id]
+      event = Event.new(name, stack.last, payload)
+      stack << event
+      event.result = yield
+      event
+    ensure
+      event.finish!
+      stack.delete(event)
+      @listeners.each { |s| s.push(event) }
+    end
+
+    def self.register(listener)
+      @listeners << listener
+    end
+
+    def self.unregister(listener)
+      @listeners.delete(listener)
+    end
+
+    class Event
+      attr_reader :name, :time, :duration, :parent, :thread_id, :payload
+      attr_accessor :result
+
+      def initialize(name, parent=nil, payload=nil)
+        @name      = name
+        @time      = Time.now
+        @thread_id = Thread.current.object_id
+        @parent    = parent
+        @payload   = payload
+      end
+
+      def finish!
+        @duration = 1000 * (Time.now.to_f - @time.to_f)
+      end
+    end
+
+    class Listener
+      attr_reader :mutex, :signaler, :thread
+
+      def initialize
+        @mutex, @signaler = Mutex.new, ConditionVariable.new
+        @stream = []
+        @thread = Thread.new do
+          loop do
+            (event = @stream.shift) ? consume(event) : wait
+          end
+        end
+      end
+
+      def wait
+        @mutex.synchronize do
+          @signaler.wait(@mutex)
+        end
+      end
+
+      def push(event)
+        @mutex.synchronize do
+          @stream.push(event)
+          @signaler.broadcast
+        end
+      end
+
+      def consume(event)
+        raise NotImplementedError
+      end
+    end
+  end
+end

activesupport/test/orchestra_test.rb

+require 'abstract_unit'
+
+class OrchestraEventTest < Test::Unit::TestCase
+  def setup
+    @parent = ActiveSupport::Orchestra::Event.new(:parent)
+  end
+
+  def test_initialization_with_name_and_parent_and_payload
+    event = ActiveSupport::Orchestra::Event.new(:awesome, @parent, :payload => "orchestra")
+    assert_equal(:awesome, event.name)
+    assert_equal(@parent, event.parent)
+    assert_equal({ :payload => "orchestra" }, event.payload)
+  end
+
+  def test_thread_id_is_set_on_initialization
+    event = ActiveSupport::Orchestra::Event.new(:awesome)
+    assert_equal Thread.current.object_id, event.thread_id
+  end
+
+  def test_current_time_is_set_on_initialization
+    previous_time = Time.now.utc
+    event = ActiveSupport::Orchestra::Event.new(:awesome)
+    assert_kind_of Time, event.time
+    assert event.time.to_f >= previous_time.to_f
+  end
+ 
+  def test_duration_is_set_when_event_finishes
+    event = ActiveSupport::Orchestra::Event.new(:awesome)
+    sleep(0.1)
+    event.finish!
+    assert_in_delta 100, event.duration, 30
+  end
+end
+
+class OrchestraMainTest < Test::Unit::TestCase
+  def setup
+    @listener = []
+    ActiveSupport::Orchestra.register @listener
+  end
+
+  def teardown
+    ActiveSupport::Orchestra.unregister @listener
+  end
+
+  def test_orchestra_allows_any_action_to_be_instrumented
+    event = ActiveSupport::Orchestra.instrument(:awesome, "orchestra") do
+      sleep(0.1)
+    end
+
+    assert_equal :awesome, event.name
+    assert_equal "orchestra", event.payload
+    assert_in_delta 100, event.duration, 30
+  end
+
+  def test_block_result_is_stored
+    event = ActiveSupport::Orchestra.instrument(:awesome, "orchestra") do
+      1 + 1
+    end
+
+    assert_equal 2, event.result
+  end
+
+  def test_events_are_published_to_a_listener
+    event = ActiveSupport::Orchestra.instrument(:awesome, "orchestra") do
+      1 + 1
+    end
+
+    assert_equal 1, @listener.size
+    assert_equal :awesome, @listener.last.name
+    assert_equal "orchestra", @listener.last.payload
+  end
+
+  def test_nested_events_can_be_instrumented
+    ActiveSupport::Orchestra.instrument(:awesome, "orchestra") do
+      ActiveSupport::Orchestra.instrument(:wot, "child") do
+        sleep(0.1)
+      end
+
+      assert_equal 1, @listener.size
+      assert_equal :wot, @listener.first.name
+      assert_equal "child", @listener.first.payload
+
+      assert_nil @listener.first.parent.duration
+      assert_in_delta 100, @listener.first.duration, 30
+    end
+
+    assert_equal 2, @listener.size
+    assert_equal :awesome, @listener.last.name
+    assert_equal "orchestra", @listener.last.payload
+    assert_in_delta 100, @listener.first.parent.duration, 30
+  end
+
+  def test_event_is_pushed_even_if_block_fails
+    ActiveSupport::Orchestra.instrument(:awesome, "orchestra") do
+      raise "OMG"
+    end rescue RuntimeError
+
+    assert_equal 1, @listener.size
+    assert_equal :awesome, @listener.last.name
+    assert_equal "orchestra", @listener.last.payload
+  end
+end
+
+class OrchestraListenerTest < Test::Unit::TestCase
+  class MyListener < ActiveSupport::Orchestra::Listener
+    attr_reader :consumed
+
+    def consume(event)
+      @consumed ||= []
+      @consumed << event
+    end
+  end
+
+  def setup
+    @listener = MyListener.new
+    ActiveSupport::Orchestra.register @listener
+  end
+
+  def teardown
+    ActiveSupport::Orchestra.unregister @listener
+  end
+
+  def test_thread_is_exposed_by_listener
+    assert_kind_of Thread, @listener.thread
+  end
+
+  def test_event_is_consumed_when_an_action_is_instrumented
+    ActiveSupport::Orchestra.instrument(:sum) do
+      1 + 1
+    end
+    sleep 0.1
+    assert_equal 1, @listener.consumed.size
+    assert_equal :sum, @listener.consumed.first.name
+    assert_equal 2, @listener.consumed.first.result
+  end
+
+  def test_with_sevaral_consumers_and_several_events
+    @another = MyListener.new
+    ActiveSupport::Orchestra.register @another
+
+    1.upto(100) do |i|
+      ActiveSupport::Orchestra.instrument(:value) do
+        i
+      end
+    end
+
+    sleep 0.1
+
+    assert_equal 100, @listener.consumed.size
+    assert_equal :value, @listener.consumed.first.name
+    assert_equal 1, @listener.consumed.first.result
+    assert_equal 100, @listener.consumed.last.result
+
+    assert_equal 100, @another.consumed.size
+    assert_equal :value, @another.consumed.first.name
+    assert_equal 1, @another.consumed.first.result
+    assert_equal 100, @another.consumed.last.result
+  ensure
+    ActiveSupport::Orchestra.unregister @another
+  end
+end