Added observation and better exception handling to SimpleActorRef.

Author: jdantonio

lib/concurrent/actor_context.rb

@@ -20,7 +20,7 @@ class << base
 
         def spawn(opts = {})
           args = opts.fetch(:args, [])
-          Concurrent::SimpleActorRef.new(self.new(*args))
+          Concurrent::SimpleActorRef.new(self.new(*args), opts)
         end
       end
     end

lib/concurrent/actor_ref.rb

@@ -1,3 +1,4 @@
+require 'concurrent/copy_on_notify_observer_set'
 require 'concurrent/utilities'
 
 module Concurrent
@@ -24,5 +25,23 @@ def <<(message)
       post(*message)
       self
     end
+
+    def add_observer(*args)
+      @observers.add_observer(*args)
+    end
+
+    def delete_observer(*args)
+      @observers.delete_observer(*args)
+    end
+
+    def delete_observers
+      @observers.delete_observers
+    end
+
+    protected
+
+    def observers
+      @observers ||= CopyOnNotifyObserverSet.new
+    end
   end
 end

lib/concurrent/simple_actor_ref.rb

@@ -8,12 +8,16 @@ module Concurrent
   class SimpleActorRef
     include ActorRef
 
-    def initialize(actor)
+    def initialize(actor, opts = {})
       @actor = actor
       @mutex = Mutex.new
       @queue = Queue.new
       @thread = nil
       @stopped = false
+      @abort_on_exception = opts.fetch(:abort_on_exception, true)
+      @reset_on_error = opts.fetch(:reset_on_error, true)
+      @exception_class = opts.fetch(:rescue_exception, false) ? Exception : StandardError
+      @observers = CopyOnNotifyObserverSet.new
     end
 
     def running?
@@ -72,7 +76,7 @@ def supervise
 
     def new_worker_thread
       Thread.new do
-        Thread.current.abort_on_exception = true
+        Thread.current.abort_on_exception = @abort_on_exception
         run_message_loop
       end
     end
@@ -84,11 +88,13 @@ def run_message_loop
 
         begin
           result = @actor.receive(*message.payload)
-        rescue => ex
-          # suppress
+        rescue @exception_class => ex
+          @actor.on_reset if @reset_on_error
         ensure
+          now = Time.now
           message.ivar.complete(ex.nil?, result, ex)
-          message.callback.call(Time.now, result, ex) if message.callback
+          message.callback.call(now, result, ex) if message.callback
+          observers.notify_observers(now, message.payload, result, ex)
         end
       end
     end

spec/concurrent/actor_ref_shared.rb

@@ -3,14 +3,19 @@
 def shared_actor_test_class
   Class.new do
     include Concurrent::ActorContext
+
     attr_reader :argv
+
     def initialize(*args)
       @argv = args
     end
+
     def receive(*msg)
       case msg.first
       when :poison
         raise StandardError
+      when :bullet
+        raise Exception
       when :terminate
         Thread.current.kill
       when :sleep
@@ -163,4 +168,57 @@ def receive(*msg)
       }.to raise_error(StandardError)
     end
   end
+
+  context 'observation' do
+
+    let(:observer_class) do
+      Class.new do
+        attr_reader :time, :msg, :value, :reason
+        def update(time, msg, value, reason)
+          @msg = msg
+          @time = time
+          @value = value
+          @reason = reason
+        end
+      end
+    end
+
+    it 'notifies observers' do
+      o1 = observer_class.new
+      o2 = observer_class.new
+
+      subject.add_observer(o1)
+      subject.add_observer(o2)
+
+      subject << :foo
+      sleep(0.1)
+
+      o1.value.should eq :foo
+      o1.reason.should be_nil
+
+      o2.value.should eq :foo
+      o2.reason.should be_nil
+    end
+
+    it 'does not notify removed observers' do
+      o1 = observer_class.new
+      o2 = observer_class.new
+
+      subject.add_observer(o1)
+      subject.add_observer(o2)
+
+      subject << :foo
+      sleep(0.1)
+
+      subject.delete_observer(o1)
+      subject << :bar
+      sleep(0.1)
+      o1.value.should_not eq :bar
+
+      subject.delete_observers
+      subject << :baz
+      sleep(0.1)
+      o1.value.should_not eq :baz
+    end
+  end
 end

spec/concurrent/simple_actor_ref_spec.rb

@@ -24,11 +24,18 @@ module Concurrent
         actor.argv.should be_empty
       end
 
-      it 'passes all :args option to the constructor' do
+      it 'passes all :args option to the actor constructor' do
         subject = shared_actor_test_class.spawn(args: [1, 2, 3, 4])
         actor = subject.instance_variable_get(:@actor)
         actor.argv.should eq [1, 2, 3, 4]
       end
+
+      it 'passes the options hash to the ActorRef constructor' do
+        subject # prevent the after(:all) block from breaking this test
+        opts = {foo: :bar, hello: :world}
+        described_class.should_receive(:new).once.with(anything, opts)
+        shared_actor_test_class.spawn(opts)
+      end
     end
 
     context 'supervision' do
@@ -82,7 +89,7 @@ module Concurrent
         subject << :foo
       end
 
-      it 'calls #on_reset when the thread is reseted' do
+      it 'calls #on_reset when the thread is started after the first time' do
         actor = subject.instance_variable_get(:@actor)
         actor.should_receive(:on_reset).once.with(no_args)
         subject << :terminate
@@ -91,6 +98,112 @@ module Concurrent
       end
     end
 
+    context 'abort_on_exception' do
+
+      after(:each) do
+        @ref.shutdown if @ref
+      end
+
+      it 'gets set on the actor thread' do
+        @ref = shared_actor_test_class.spawn(abort_on_exception: true)
+        @ref << :foo
+        sleep(0.1)
+        @ref.instance_variable_get(:@thread).abort_on_exception.should be_true
+
+        @ref = shared_actor_test_class.spawn(abort_on_exception: false)
+        @ref << :foo
+        sleep(0.1)
+        @ref.instance_variable_get(:@thread).abort_on_exception.should be_false
+      end
+
+      it 'defaults to true' do
+        @ref = shared_actor_test_class.spawn
+        @ref << :foo
+        sleep(0.1)
+        @ref.instance_variable_get(:@thread).abort_on_exception.should be_true
+      end
+    end
+
+    context 'reset_on_error' do
+
+      after(:each) do
+        @ref.shutdown if @ref
+      end
+
+      it 'causes #on_reset to be called on exception when true' do
+        @ref = shared_actor_test_class.spawn(reset_on_error: true)
+        actor = @ref.instance_variable_get(:@actor)
+        actor.should_receive(:on_reset).once.with(no_args)
+        @ref << :poison
+        sleep(0.1)
+      end
+
+      it 'prevents #on_reset form being called on exception when false' do
+        @ref = shared_actor_test_class.spawn(reset_on_error: false)
+        actor = @ref.instance_variable_get(:@actor)
+        actor.should_not_receive(:on_reset).with(any_args)
+        @ref << :poison
+        sleep(0.1)
+      end
+
+      it 'defaults to true' do
+        @ref = shared_actor_test_class.spawn
+        actor = @ref.instance_variable_get(:@actor)
+        actor.should_receive(:on_reset).once.with(no_args)
+        @ref << :poison
+        sleep(0.1)
+      end
+    end
+
+    context 'rescue_exception' do
+
+      after(:each) do
+        @ref.shutdown if @ref
+      end
+
+      it 'rescues Exception in the actor thread when true' do
+        @ref = shared_actor_test_class.spawn(
+          abort_on_exception: false,
+          rescue_exception: true
+        )
+
+        ivar = @ref.post(:poison)
+        sleep(0.1)
+        ivar.reason.should be_a StandardError
+
+        ivar = @ref.post(:bullet)
+        sleep(0.1)
+        ivar.reason.should be_a Exception
+      end
+
+      it 'rescues StandardError in the actor thread when false' do
+        @ref = shared_actor_test_class.spawn(
+          abort_on_exception: false,
+          rescue_exception: false
+        )
+
+        ivar = @ref.post(:poison)
+        sleep(0.1)
+        ivar.reason.should be_a StandardError
+
+        ivar = @ref.post(:bullet)
+        sleep(0.1)
+        ivar.reason.should be_nil
+      end
+
+      it 'defaults to false' do
+        @ref = shared_actor_test_class.spawn(abort_on_exception: false)
+
+        ivar = @ref.post(:poison)
+        sleep(0.1)
+        ivar.reason.should be_a StandardError
+
+        ivar = @ref.post(:bullet)
+        sleep(0.1)
+        ivar.reason.should be_nil
+      end
+    end
+
     context '#shutdown' do
 
       it 'calls #on_shutdown when shutdown' do