Added #join method to SimpleActorRef and shared specs.

Author: jdantonio

lib/concurrent/simple_actor_ref.rb

@@ -1,6 +1,7 @@
 require 'thread'
 
 require 'concurrent/actor_ref'
+require 'concurrent/event'
 require 'concurrent/ivar'
 
 module Concurrent
@@ -13,25 +14,25 @@ def initialize(actor, opts = {})
       @mutex = Mutex.new
       @queue = Queue.new
       @thread = nil
-      @stopped = false
+      @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?
-      @mutex.synchronize{ @stopped == false }
+      ! @stop_event.set?
     end
 
     def shutdown?
-      @mutex.synchronize{ @stopped == true }
+      @stop_event.set?
     end
 
     def post(*msg, &block)
       raise ArgumentError.new('message cannot be empty') if msg.empty?
       @mutex.synchronize do
-        supervise unless @stopped == true
+        supervise unless shutdown?
       end
       ivar = IVar.new
       @queue.push(Message.new(msg, ivar, block))
@@ -52,15 +53,19 @@ def post!(seconds, *msg)
 
     def shutdown
       @mutex.synchronize do
-        return if @stopped
-        @stopped = true
+        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)

spec/concurrent/actor_ref_shared.rb

@@ -169,6 +169,57 @@ def receive(*msg)
     end
   end
 
+  context '#join' do
+
+    it 'blocks until shutdown when no limit is given' do
+      start = Time.now
+      subject << :foo # start the actor's thread
+      Thread.new{ sleep(1); subject.shutdown }
+      subject.join
+      stop = Time.now
+
+      subject.should be_shutdown
+      stop.should >= start + 1
+      stop.should <= start + 2
+    end
+
+    it 'blocks for no more than the given number of seconds' do
+      start = Time.now
+      subject << :foo # start the actor's thread
+      Thread.new{ sleep(5); subject.shutdown }
+      subject.join(1)
+      stop = Time.now
+
+      stop.should >= start + 1
+      stop.should <= start + 2
+    end
+
+    it 'returns true when shutdown has completed before timeout' do
+      subject << :foo # start the actor's thread
+      Thread.new{ sleep(1); subject.shutdown }
+      subject.join.should be_true
+    end
+
+    it 'returns false on timeout' do
+      subject << :foo # start the actor's thread
+      Thread.new{ sleep(5); subject.shutdown }
+      subject.join(1).should be_false
+    end
+
+    it 'returns immediately when already shutdown' do
+      start = Time.now
+      subject << :foo # start the actor's thread
+      sleep(0.1)
+      subject.shutdown
+      sleep(0.1)
+
+      start = Time.now
+      subject.join
+      Time.now.should >= start
+      Time.now.should <= start + 0.1
+    end
+  end
+
   context 'observation' do
 
     let(:observer_class) do