Class: Concurrent::Actor::AbstractContext Abstract

Inherits:
Object
  • Object
show all
Includes:
InternalDelegations, TypeCheck
Defined in:
lib/concurrent-ruby-edge/concurrent/actor/context.rb

Overview

This class is abstract.

implement #on_message and #behaviour_definition

New actor is defined by subclassing RestartingContext, Context and defining its abstract methods. AbstractContext can be subclassed directly to implement more specific behaviour see Root implementation.

Example of ac actor definition:

Message = Struct.new :action, :value 

class AnActor < Concurrent::Actor::RestartingContext
  def initialize(init)
    @counter = init
  end

  # override #on_message to define actor's behaviour on message received
  def on_message(message)
    case message.action
    when :add
      @counter = @counter + message.value
    when :subtract
      @counter = @counter - message.value
    when :value
      @counter
    else
      pass
    end
  end

  # set counter to zero when there is an error
  def on_event(event)
    if event == :reset
      @counter = 0 # ignore initial value
    end
  end
end 

an_actor = AnActor.spawn name: 'an_actor', args: 10 
an_actor << Message.new(:add, 1) << Message.new(:subtract, 2) 
an_actor.ask!(Message.new(:value, nil))            # => 9
an_actor << :boo << Message.new(:add, 1) 
an_actor.ask!(Message.new(:value, nil))            # => 1
an_actor << :terminate!
    # => #

See methods of AbstractContext what else can be tweaked, e.g #default_reference_class

Direct Known Subclasses

Context, RestartingContext, Root

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#core ⇒ undocumented (readonly) 



30
31
32
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 30

def core
  @core
end

Class Method Details

.spawn(name_or_opts, *args, &block) ⇒ undocumented

Behaves as Concurrent::Actor.spawn but :class is auto-inserted based on receiver so it can be omitted.

Examples:

by class and name

AdHoc.spawn(:ping1) { -> message { message } }

by option hash

inc2 = AdHoc.spawn(name:     'increment by 2',
                   args:     [2],
                   executor: Concurrent.configuration.global_task_pool) do |increment_by|
  lambda { |number| number + increment_by }
end
inc2.ask!(2) # => 4

See Also:



117
118
119
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 117

def self.spawn(name_or_opts, *args, &block)
  Actor.spawn to_spawn_options(name_or_opts, *args), &block
end

.spawn!(name_or_opts, *args, &block) ⇒ undocumented

behaves as Concurrent::Actor.spawn! but :class is auto-inserted based on receiver so it can be omitted.



122
123
124
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 122

def self.spawn!(name_or_opts, *args, &block)
  Actor.spawn! to_spawn_options(name_or_opts, *args), &block
end

Instance Method Details

#ask(message) ⇒ undocumented Also known as: ask! 



98
99
100
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 98

def ask(message)
  raise 'actor cannot ask itself'
end

#behaviour_definition ⇒ Array<Array(Behavior::Abstract, Array<Object>)>

Returns:

Raises:

  • (NotImplementedError) 


72
73
74
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 72

def behaviour_definition
  raise NotImplementedError
end

#dead_letter_routing ⇒ Reference

Defines an actor responsible for dead letters. Any rejected message send with Reference#tell is sent there, a message with future is considered already monitored for failures. Default behaviour is to use #dead_letter_routing of the parent, so if no #dead_letter_routing method is overridden in parent-chain the message ends up in Actor.root.dead_letter_routing agent which will log warning.

Returns:



67
68
69
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 67

def dead_letter_routing
  parent.dead_letter_routing
end

#default_executor ⇒ Executor

override to se different default executor, e.g. to change it to global_operation_pool

Returns:

  • (Executor) 


89
90
91
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 89

def default_executor
  Concurrent.global_io_executor
end

#default_reference_class ⇒ CLass

override if different class for reference is needed

Returns:



83
84
85
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 83

def default_reference_class
  Reference
end

#envelope ⇒ Envelope

Returns current envelope, accessible inside #on_message processing.

Returns:

  • (Envelope) —

    current envelope, accessible inside #on_message processing



77
78
79
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 77

def envelope
  @envelope or raise 'envelope not set'
end

#on_envelope(envelope) ⇒ undocumented

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



46
47
48
49
50
51
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 46

def on_envelope(envelope)
  @envelope = envelope
  on_message envelope.message
ensure
  @envelope = nil
end

#on_event(event) ⇒ undocumented

override to add custom code invocation on internal events like :terminated, :resumed, anError.



42
43
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 42

def on_event(event)
end

#on_message(message) ⇒ Object

This method is abstract.

override to define Actor's behaviour

Note:

self should not be returned (or sent to other actors), PublicDelegations#reference should be used instead

Returns a result which will be used to set the Future supplied to Reference#ask.

Parameters:

  • message (Object)

Returns:

  • (Object) —

    a result which will be used to set the Future supplied to Reference#ask

Raises:

  • (NotImplementedError) 


37
38
39
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 37

def on_message(message)
  raise NotImplementedError
end

#pass ⇒ undocumented

if you want to pass the message to next behaviour, usually Behaviour::ErrorsOnUnknownMessage



55
56
57
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 55

def pass
  core.behaviour!(Behaviour::ExecutesContext).pass envelope
end

#tell(message) ⇒ undocumented Also known as: <<

tell self a message



94
95
96
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/context.rb', line 94

def tell(message)
  reference.tell message
end

#behaviour(behaviour_class) ⇒ undocumented Originally defined in module InternalDelegations

see Core#behaviour

#behaviour!(behaviour_class) ⇒ undocumented Originally defined in module InternalDelegations

see Core#behaviour!

#Child!(value, *types) ⇒ undocumented Originally defined in module TypeCheck

#Child?(value, *types) ⇒ Boolean Originally defined in module TypeCheck

Returns:

  • (Boolean)

#children ⇒ undocumented Originally defined in module InternalDelegations

See Also:

#context ⇒ AbstractContext Originally defined in module InternalDelegations

Returns:

#context_class ⇒ undocumented Also known as: actor_class Originally defined in module PublicDelegations

See Also:

#executor ⇒ undocumented Originally defined in module PublicDelegations

See Also:

#log(level, message = nil, &block) ⇒ undocumented Originally defined in module InternalDelegations

delegates to core.log

See Also:

  • Logging#log

#Match!(value, *types) ⇒ undocumented Originally defined in module TypeCheck

#Match?(value, *types) ⇒ Boolean Originally defined in module TypeCheck

Returns:

  • (Boolean)

#name ⇒ undocumented Originally defined in module PublicDelegations

See Also:

#parent ⇒ undocumented Originally defined in module PublicDelegations

See Also:

#path ⇒ undocumented Originally defined in module PublicDelegations

See Also:

#redirect(reference, envelope = self.envelope) ⇒ undocumented Originally defined in module InternalDelegations

#reference ⇒ undocumented Also known as: ref Originally defined in module PublicDelegations

See Also:

#terminate!(reason = nil) ⇒ undocumented Originally defined in module InternalDelegations

See Also:

  • Termination#terminate!

#terminated? ⇒ Boolean Originally defined in module InternalDelegations

Returns:

  • (Boolean)

See Also:

  • Termination#terminated?

#Type!(value, *types) ⇒ undocumented Originally defined in module TypeCheck

#Type?(value, *types) ⇒ Boolean Originally defined in module TypeCheck

Returns:

  • (Boolean)