Class: Concurrent::Actor::Behaviour::Buffer

Inherits:
Abstract
  • Object
show all
Defined in:
lib/concurrent-ruby-edge/concurrent/actor/behaviour/buffer.rb

Overview

Any message reaching this behaviour is buffered. Only one message is is scheduled at any given time. Others are kept in buffer until another one can be scheduled. This effectively means that messages handled by behaviours before buffer have higher priority and they can be processed before messages arriving into buffer. This allows for the processing of internal actor messages like (‘:link`, `:supervise`) first.

Instance Attribute Summary

Attributes inherited from Abstract

#core, #subsequent

Instance Method Summary collapse

Methods inherited from Abstract

#broadcast, #pass, #reject_envelope

Methods included from InternalDelegations

#behaviour, #behaviour!, #children, #context, #dead_letter_routing, #log, #redirect, #terminate!, #terminated?

Methods included from PublicDelegations

#context_class, #executor, #name, #parent, #path, #reference

Methods included from TypeCheck

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

Constructor Details

#initialize(core, subsequent, core_options) ⇒ Buffer

Returns a new instance of Buffer.



14
15
16
17
18
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/buffer.rb', line 14

def initialize(core, subsequent, core_options)
  super core, subsequent, core_options
  @buffer                     = []
  @receive_envelope_scheduled = false
end

Instance Method Details

#on_envelope(envelope) ⇒ Object 



20
21
22
23
24
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/buffer.rb', line 20

def on_envelope(envelope)
  @buffer.push envelope
  process_envelopes?
  MESSAGE_PROCESSED
end

#on_event(public, event) ⇒ Object 



46
47
48
49
50
51
52
53
54
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/buffer.rb', line 46

def on_event(public, event)
  event_name, _ = event
  case event_name
  when :terminated, :restarted
    @buffer.each { |envelope| reject_envelope envelope }
    @buffer.clear
  end
  super public, event_name
end

#process_envelope ⇒ Object 



37
38
39
40
41
42
43
44
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/buffer.rb', line 37

def process_envelope
  envelope = @buffer.shift
  return nil unless envelope
  pass envelope
ensure
  @receive_envelope_scheduled = false
  core.schedule_execution { process_envelopes? }
end

#process_envelopes? ⇒ Boolean

Ensures that only one envelope processing is scheduled with #schedule_execution, this allows other scheduled blocks to be executed before next envelope processing. Simply put this ensures that Core is still responsive to internal calls (like add_child) even though the Actor is flooded with messages.

Returns:

  • (Boolean) 


30
31
32
33
34
35
 
# File 'lib/concurrent-ruby-edge/concurrent/actor/behaviour/buffer.rb', line 30

def process_envelopes?
  unless @buffer.empty? || @receive_envelope_scheduled
    @receive_envelope_scheduled = true
    process_envelope
  end
end