123456789_123456789_123456789_123456789_123456789_

Class: Concurrent::Actor::Behaviour::Pausing

Relationships & Source Files
Super Chains via Extension / Inclusion / Inheritance
Class Chain:
self, Abstract
Instance Chain:
self, Abstract, Actor::InternalDelegations, Concern::Logging, Actor::PublicDelegations, Actor::TypeCheck
Inherits: Concurrent::Actor::Behaviour::Abstract
Defined in: lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb

Overview

Note:

TODO missing example

Allows to pause actors on errors. When paused all arriving messages are collected and processed after the actor is resumed or reset. Resume will simply continue with next message. Reset also reinitialized context.

Constant Summary

Concern::Logging - Included

SEV_LABEL

Class Method Summary

Abstract - Inherited

.new

Instance Attribute Summary

Abstract - Inherited

#core, #subsequent

Actor::InternalDelegations - Included

#terminated?

Instance Method Summary

Abstract - Inherited

#broadcast

broadcasts event to all behaviours and context.
#on_envelope

override to add extra behaviour.
#on_event

override to add extra behaviour.
#pass, #reject_envelope

Actor::InternalDelegations - Included

#behaviour

see Core#behaviour
#behaviour!

see Core#behaviour!
#children, #context, #dead_letter_routing,
#log

delegates to core.log.
#redirect, #terminate!

Concern::Logging - Included

#log

Logs through Concurrent.global_logger, it can be overridden by setting @logger.

Actor::PublicDelegations - Included

#actor_class

Alias for PublicDelegations#context_class.
#context_class, #executor, #name, #parent, #path,
#ref

Alias for PublicDelegations#reference.
#reference

Actor::TypeCheck - Included

#Child!, #Child?, #Match!, #Match?, #Type!, #Type?

Constructor Details

.new(core, subsequent, core_options) ⇒ Pausing

[ GitHub ] 

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 13



def initialize(core, subsequent, core_options)
  super core, subsequent, core_options
  @paused   = false
  @deferred = []
end


  







Instance Attribute Details



  

    #paused?  ⇒ Boolean  (readonly)
  



  

    
  





  


  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 19



def paused?
  @paused
end


  







Instance Method Details



  

    #do_pause   (private)
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 82



def do_pause
  @paused = true
  nil
end


  








  

    #do_reset   (private)
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 93



def do_reset
  rebuild_context
  do_resume
  reschedule_deferred
  nil
end


  








  

    #do_restart   (private)
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 100



def do_restart
  rebuild_context
  reject_deferred
  do_resume
  nil
end


  








  

    #do_resume   (private)
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 87



def do_resume
  @paused = false
  reschedule_deferred
  nil
end


  








  

    #on_envelope(envelope)  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 23



def on_envelope(envelope)
  case envelope.message
  when :pause!
    pause!
  when :paused?
    paused?
  when :resume!
    resume!
  when :reset!
    reset!
  when :restart!
    restart!
  else
    if paused?
      @deferred << envelope
      MESSAGE_PROCESSED
    else
      pass envelope
    end
  end
end


  








  

    #on_event(public, event)  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 74



def on_event(public, event)
  event_name, _ = event
  reject_deferred if event_name == :terminated
  super public, event
end


  








  

    #pause!(error = nil)  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 45



def pause!(error = nil)
  do_pause
  broadcast true, error || :paused
  true
end


  








  

    #rebuild_context   (private)
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 107



def rebuild_context
  core.allocate_context
  core.build_context
  nil
end


  








  

    #reject_deferred   (private)
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 118



def reject_deferred
  @deferred.each { |envelope| reject_envelope envelope }
  @deferred.clear
end


  








  

    #reschedule_deferred   (private)
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 113



def reschedule_deferred
  @deferred.each { |envelope| core.schedule_execution { core.process_envelope envelope } }
  @deferred.clear
end


  








  

    #reset!  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 58



def reset!
  return false unless paused?
  broadcast(false, :resetting)
  do_reset
  broadcast(true, :reset)
  true
end


  








  

    #restart!  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 66



def restart!
  return false unless paused?
  broadcast(false, :restarting)
  do_restart
  broadcast(true, :restarted)
  true
end


  








  

    #resume!  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/pausing.rb', line 51



def resume!
  return false unless paused?
  do_resume
  broadcast(true, :resumed)
  true
end