Merge pull request #48 from jdantonio/actor-context

Initial implementation of ActorContext and the #spawn function

Author: jdantonio

lib/concurrent.rb

@@ -32,6 +32,9 @@
 require 'concurrent/tvar'
 require 'concurrent/utilities'
 
+require 'concurrent/actor_context'
+require 'concurrent/simple_actor_ref'
+
 require 'concurrent/cached_thread_pool'
 require 'concurrent/fixed_thread_pool'
 require 'concurrent/immediate_executor'

lib/concurrent/actor.rb

@@ -179,6 +179,7 @@ def initialize(queue)
     #
     # @deprecated +Actor+ is being replaced with a completely new framework prior to v1.0.0
     def self.pool(count, *args, &block)
+      warn '[DEPRECATED] `Actor` is deprecated and will be replaced with `ActorContext`.'
       raise ArgumentError.new('count must be greater than zero') unless count > 0
       mailbox = Queue.new
       actors = count.times.collect do
@@ -211,13 +212,15 @@ def self.pool(count, *args, &block)
     # 
     # @!visibility public
     def act(*message)
+      warn '[DEPRECATED] `Actor` is deprecated and will be replaced with `ActorContext`.'
       raise NotImplementedError.new("#{self.class} does not implement #act")
     end
 
     # @!visibility private
     #
     # @deprecated +Actor+ is being replaced with a completely new framework prior to v1.0.0
     def on_run # :nodoc:
+      warn '[DEPRECATED] `Actor` is deprecated and will be replaced with `ActorContext`.'
       queue.clear
     end
 

lib/concurrent/actor_context.rb

@@ -0,0 +1,28 @@
+require 'concurrent/simple_actor_ref'
+
+module Concurrent
+
+  module ActorContext
+
+    def on_start
+    end
+
+    def on_reset
+    end
+
+    def on_shutdown
+    end
+
+    def self.included(base)
+
+      class << base
+        protected :new
+
+        def spawn(opts = {})
+          args = opts.fetch(:args, [])
+          Concurrent::SimpleActorRef.new(self.new(*args), opts)
+        end
+      end
+    end
+  end
+end

lib/concurrent/actor_ref.rb

@@ -0,0 +1,39 @@
+require 'concurrent/copy_on_notify_observer_set'
+
+module Concurrent
+
+  module ActorRef
+
+    #NOTE: Required API methods
+    #      Must be implemented in all subclasses
+    #def post(*msg, &block)
+    #def post!(*msg)
+    #def running?
+    #def shutdown?
+    #def shutdown
+    #def join(timeout = nil)
+
+    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

@@ -0,0 +1,108 @@
+require 'thread'
+
+require 'concurrent/actor_ref'
+require 'concurrent/event'
+require 'concurrent/ivar'
+
+module Concurrent
+
+  class SimpleActorRef
+    include ActorRef
+
+    def initialize(actor, opts = {})
+      @actor = actor
+      @mutex = Mutex.new
+      @queue = Queue.new
+      @thread = nil
+      @stop_event = Event.new
+      @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?
+      ! @stop_event.set?
+    end
+
+    def shutdown?
+      @stop_event.set?
+    end
+
+    def post(*msg, &block)
+      raise ArgumentError.new('message cannot be empty') if msg.empty?
+      @mutex.synchronize do
+        supervise unless shutdown?
+      end
+      ivar = IVar.new
+      @queue.push(Message.new(msg, ivar, block))
+      ivar
+    end
+
+    def post!(seconds, *msg)
+      raise Concurrent::TimeoutError if seconds == 0
+      ivar = self.post(*msg)
+      ivar.value(seconds)
+      if ivar.incomplete?
+        raise Concurrent::TimeoutError
+      elsif ivar.reason
+        raise ivar.reason
+      end
+      ivar.value
+    end
+
+    def shutdown
+      @mutex.synchronize do
+        return if shutdown?
+        if @thread && @thread.alive?
+          @thread.kill 
+          @actor.on_shutdown
+        end
+        @stop_event.set
+      end
+    end
+
+    def join(timeout = nil)
+      @stop_event.wait(timeout)
+    end
+
+    private
+
+    Message = Struct.new(:payload, :ivar, :callback)
+
+    def supervise
+      if @thread.nil?
+        @actor.on_start
+        @thread = new_worker_thread
+      elsif ! @thread.alive?
+        @actor.on_reset
+        @thread = new_worker_thread
+      end
+    end
+
+    def new_worker_thread
+      Thread.new do
+        Thread.current.abort_on_exception = @abort_on_exception
+        run_message_loop
+      end
+    end
+
+    def run_message_loop
+      loop do
+        message = @queue.pop
+        result = ex = nil
+
+        begin
+          result = @actor.receive(*message.payload)
+        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(now, result, ex) if message.callback
+          observers.notify_observers(now, message.payload, result, ex)
+        end
+      end
+    end
+  end
+end

spec/concurrent/actor_context_spec.rb

@@ -0,0 +1,37 @@
+require 'spec_helper'
+
+module Concurrent
+
+  describe ActorContext do
+
+    let(:described_class) do
+      Class.new do
+        include ActorContext
+      end
+    end
+
+    it 'protects #initialize' do
+      expect {
+        described_class.new
+      }.to raise_error(NoMethodError)
+    end
+
+    context 'callbacks' do
+
+      subject { described_class.send(:new) }
+
+      specify { subject.should respond_to :on_start }
+
+      specify { subject.should respond_to :on_reset }
+
+      specify { subject.should respond_to :on_shutdown }
+    end
+
+    context '#spawn' do
+
+      it 'returns an ActorRef' do
+        described_class.spawn.should be_a ActorRef
+      end
+    end
+  end
+end