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 Method Summary collapse

Constructor Details

#initialize(core, subsequent, core_options) ⇒ Buffer

Returns a new instance of Buffer.

[View source]

12
13
14
15
16
 
# File 'lib/concurrent-ruby-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

[View source]

18
19
20
21
22
 
# File 'lib/concurrent-ruby-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

[View source]

44
45
46
47
48
49
50
51
52
 
# File 'lib/concurrent-ruby-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

[View source]

35
36
37
38
39
40
41
42
 
# File 'lib/concurrent-ruby-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)
[View source]

28
29
30
31
32
33
 
# File 'lib/concurrent-ruby-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