Add :restart! directive

as :reset but it also drops all messages

Author: pitr-ch

lib/concurrent/actor/behaviour/buffer.rb

@@ -41,7 +41,8 @@ def process_envelope
         end
 
         def on_event(event)
-          if event == :terminated
+          case event
+          when :terminated, :restarted
             @buffer.each { |envelope| reject_envelope envelope }
             @buffer.clear
           end

lib/concurrent/actor/behaviour/pausing.rb

@@ -22,8 +22,8 @@ def on_envelope(envelope)
             from_supervisor?(envelope) { resume! }
           when :reset!
             from_supervisor?(envelope) { reset! }
-            # when :restart! TODO
-            #   from_supervisor?(envelope) { reset! }
+          when :restart!
+            from_supervisor?(envelope) { restart! }
           else
             if @paused
               @buffer << envelope
@@ -34,40 +34,48 @@ def on_envelope(envelope)
           end
         end
 
+        def from_supervisor?(envelope)
+          if behaviour!(Supervised).supervisor == envelope.sender
+            yield
+          else
+            false
+          end
+        end
+
         def pause!(error = nil)
           @paused = true
           broadcast(error || :paused)
           true
         end
 
         def resume!(broadcast = true)
-          @buffer.each { |envelope| core.schedule_execution { pass envelope } }
-          @buffer.clear
           @paused = false
           broadcast(:resumed) if broadcast
           true
         end
 
-        def from_supervisor?(envelope)
-          if behaviour!(Supervised).supervisor == envelope.sender
-            yield
-          else
-            false
-          end
-        end
-
-        def reset!
+        def reset!(broadcast = true)
           core.allocate_context
           core.build_context
           resume!(false)
-          broadcast(:reset)
+          broadcast(:reset) if broadcast
+          true
+        end
+
+        def restart!
+          reset! false
+          broadcast(:restarted)
           true
         end
 
         def on_event(event)
-          if event == :terminated
+          case event
+          when :terminated, :restarted
             @buffer.each { |envelope| reject_envelope envelope }
             @buffer.clear
+          when :resumed, :reset
+            @buffer.each { |envelope| core.schedule_execution { pass envelope } }
+            @buffer.clear
           end
           super event
         end

spec/concurrent/actor_spec.rb

@@ -336,6 +336,39 @@ def on_message(message)
           expect(queue.pop).to eq :init # rebuilds context
           expect(queue.pop).to eq :reset
           terminate_actors test
+
+          queue              = Queue.new
+          resuming_behaviour = Behaviour.restarting_behaviour_definition.map do |c, args|
+            if Behaviour::Supervising == c
+              [c, [:restart!, :one_for_one]]
+            else
+              [c, args]
+            end
+          end
+
+          test = AdHoc.spawn name: :tester, behaviour_definition: resuming_behaviour do
+            actor = AdHoc.spawn name:                 :pausing,
+                                behaviour_definition: Behaviour.restarting_behaviour_definition do
+              queue << :init
+              -> m { m == :add ? 1 : pass }
+            end
+
+            actor << :supervise
+            queue << actor.ask!(:supervisor)
+            actor << nil
+            queue << actor.ask(:add)
+
+            -> m do
+              queue << m
+            end
+          end
+
+          expect(queue.pop).to eq :init
+          expect(queue.pop).to eq test
+          expect(queue.pop.wait.reason).to be_a_kind_of(ActorTerminated)
+          expect(queue.pop).to eq :init
+          expect(queue.pop).to eq :restarted
+          terminate_actors test
         end
 
       end