Improve actor events

Events can be private and public, so far only difference is that Linking will
pass to linked actors only public messages. Adding private :restarting and
:resetting events which are send before the actor restarts or resets allowing
to add callbacks to cleanup current child actors.

Author: pitr-ch

lib/concurrent/actor/behaviour/abstract.rb

@@ -25,15 +25,15 @@ def pass(envelope)
 
         # override to add extra behaviour
         # @note super needs to be called not to break the chain
-        def on_event(event)
-          subsequent.on_event event if subsequent
+        def on_event(public, event)
+          subsequent.on_event public, event if subsequent
         end
 
         # broadcasts event to all behaviours and context
         # @see #on_event
         # @see AbstractContext#on_event
-        def broadcast(event)
-          core.broadcast(event)
+        def broadcast(public, event)
+          core.broadcast(public, event)
         end
 
         def reject_envelope(envelope)

lib/concurrent/actor/behaviour/buffer.rb

@@ -40,13 +40,13 @@ def process_envelope
           core.schedule_execution { process_envelopes? }
         end
 
-        def on_event(event)
+        def on_event(public, event)
           case event
           when :terminated, :restarted
             @buffer.each { |envelope| reject_envelope envelope }
             @buffer.clear
           end
-          super event
+          super public, event
         end
       end
     end

lib/concurrent/actor/behaviour/executes_context.rb

@@ -7,10 +7,9 @@ def on_envelope(envelope)
           context.on_envelope envelope
         end
 
-        def on_event(event)
+        def on_event(public, event)
           context.on_event(event)
-          core.log Logging::DEBUG, "event: #{event.inspect}"
-          super event
+          super public, event
         end
       end
     end

lib/concurrent/actor/behaviour/linking.rb

@@ -3,7 +3,8 @@ module Actor
     module Behaviour
 
       # Links the actor to other actors and sends actor's events to them,
-      # like: `:terminated`, `:paused`, errors, etc.
+      # like: `:terminated`, `:paused`, `:resumed`, errors, etc.
+      # Linked actor needs to handle those messages.
       #
       #     listener = AdHoc.spawn name: :listener do
       #       lambda do |message|
@@ -65,10 +66,10 @@ def unlink(ref)
           true
         end
 
-        def on_event(event)
-          @linked.each { |a| a << event }
+        def on_event(public, event)
+          @linked.each { |a| a << event } if public
           @linked.clear if event == :terminated
-          super event
+          super public, event
         end
       end
     end

lib/concurrent/actor/behaviour/pausing.rb

@@ -35,41 +35,77 @@ def on_envelope(envelope)
         end
 
         def pause!(error = nil)
-          @paused = true
-          broadcast(error || :paused)
+          do_pause
+          broadcast true, error || :paused
           true
         end
 
-        def resume!(broadcast = true)
-          @paused = false
-          broadcast(:resumed) if broadcast
+        def resume!
+          do_resume
+          broadcast(true, :resumed)
           true
         end
 
-        def reset!(broadcast = true)
-          core.allocate_context
-          core.build_context
-          resume!(false)
-          broadcast(:reset) if broadcast
+        def reset!
+          broadcast(false, :resetting)
+          do_reset
+          broadcast(true, :reset)
           true
         end
 
         def restart!
-          reset! false
-          broadcast(:restarted)
+          broadcast(false, :restarting)
+          do_restart
+          broadcast(true, :restarted)
           true
         end
 
-        def on_event(event)
-          case event
-          when :terminated, :restarted
-            @buffer.each { |envelope| reject_envelope envelope }
-            @buffer.clear
-          when :resumed, :reset
-            @buffer.each { |envelope| core.schedule_execution { core.process_envelope envelope } }
-            @buffer.clear
-          end
-          super event
+        def on_event(public, event)
+          reject_buffer if event == :terminated
+          super public, event
+        end
+
+        private
+
+        def do_pause
+          @paused = true
+          nil
+        end
+
+        def do_resume
+          @paused = false
+          reschedule_buffer
+          nil
+        end
+
+        def do_reset
+          rebuild_context
+          do_resume
+          reschedule_buffer
+          nil
+        end
+
+        def do_restart
+          rebuild_context
+          reject_buffer
+          do_resume
+          nil
+        end
+
+        def rebuild_context
+          core.allocate_context
+          core.build_context
+          nil
+        end
+
+        def reschedule_buffer
+          @buffer.each { |envelope| core.schedule_execution { core.process_envelope envelope } }
+          @buffer.clear
+        end
+
+        def reject_buffer
+          @buffer.each { |envelope| reject_envelope envelope }
+          @buffer.clear
         end
       end
     end

lib/concurrent/actor/behaviour/supervised.rb

@@ -68,9 +68,9 @@ def un_supervise(ref)
           end
         end
 
-        def on_event(event)
+        def on_event(public, event)
           @supervisor = nil if event == :terminated
-          super event
+          super public, event
         end
       end
     end

lib/concurrent/actor/behaviour/terminates_children.rb

@@ -3,10 +3,10 @@ module Actor
     module Behaviour
       # Terminates all children when the actor terminates.
       class TerminatesChildren < Abstract
-        def on_event(event)
+        def on_event(public, event)
           # TODO set event in Termination after all children are terminated, requires new non-blocking join on Future
           children.map { |ch| ch << :terminate! } if event == :terminated
-          super event
+          super public, event
         end
       end
     end

lib/concurrent/actor/behaviour/termination.rb

@@ -28,7 +28,7 @@ def on_envelope(envelope)
             terminated?
           when :terminate!
             terminate!
-          when :terminated_event
+          when :terminated_event # TODO rename to :termination_event
             terminated
           else
             if terminated?
@@ -45,7 +45,7 @@ def on_envelope(envelope)
         def terminate!
           return true if terminated?
           terminated.set
-          broadcast(:terminated) # TODO do not end up in Dead Letter Router
+          broadcast(true, :terminated) # TODO do not end up in Dead Letter Router
           parent << :remove_child if parent
           true
         end

lib/concurrent/actor/core.rb

@@ -87,7 +87,7 @@ def initialize(opts = {}, &block)
               build_context
 
               messages.each do |message|
-                log DEBUG, "preprocessing #{message} from #{parent}"
+                log DEBUG, "preprocessing #{message.inspect} from #{parent}"
                 process_envelope Envelope.new(message, nil, parent, reference)
               end
 
@@ -176,8 +176,9 @@ def schedule_execution
         nil
       end
 
-      def broadcast(event)
-        @first_behaviour.on_event(event)
+      def broadcast(public, event)
+        log Logging::DEBUG, "event: #{event.inspect} (#{public ? 'public' : 'private'})"
+        @first_behaviour.on_event(public, event)
       end
 
       # @param [Class] behaviour_class