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

Inherits:
Abstract
  • Object
show all
Defined in:
lib-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 Method Summary collapse

Constructor Details

#initialize(core, subsequent, core_options) ⇒ Buffer

Returns a new instance of Buffer.



12
13
14
15
16
 
# File 'lib-edge/concurrent/actor/behaviour/buffer.rb', line 12

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

Instance Method Details

#on_envelope(envelope) ⇒ undocumented 



18
19
20
21
22
 
# File 'lib-edge/concurrent/actor/behaviour/buffer.rb', line 18

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

#on_event(public, event) ⇒ undocumented 



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

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 ⇒ undocumented 



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

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) 


28
29
30
31
32
33
 
# File 'lib-edge/concurrent/actor/behaviour/buffer.rb', line 28

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