Class: Concurrent::Actor::Utils::Balancer
- Inherits:
-
RestartingContext
- Object
- AbstractContext
- RestartingContext
- Concurrent::Actor::Utils::Balancer
- Defined in:
- lib/concurrent-ruby-edge/concurrent/actor/utils/balancer.rb
Overview
Distributes messages between subscribed actors. Each actor'll get only one message then it's unsubscribed. The actor needs to resubscribe when it's ready to receive next message. It will buffer the messages if there is no worker registered.
Instance Method Summary collapse
- #distribute ⇒ undocumented
-
#initialize ⇒ Balancer
constructor
A new instance of Balancer.
- #on_message(message) ⇒ undocumented
Constructor Details
#initialize ⇒ Balancer
Returns a new instance of Balancer.
11 12 13 14 |
# File 'lib/concurrent-ruby-edge/concurrent/actor/utils/balancer.rb', line 11 def initialize @receivers = [] @buffer = [] end |
Instance Method Details
#distribute ⇒ undocumented
35 36 37 38 39 |
# File 'lib/concurrent-ruby-edge/concurrent/actor/utils/balancer.rb', line 35 def distribute while !@receivers.empty? && !@buffer.empty? redirect @receivers.shift, @buffer.shift end end |
#on_message(message) ⇒ undocumented
16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 |
# File 'lib/concurrent-ruby-edge/concurrent/actor/utils/balancer.rb', line 16 def () command, who = case command when :subscribe @receivers << (who || envelope.sender) distribute true when :unsubscribe @receivers.delete(who || envelope.sender) true when :subscribed? @receivers.include?(who || envelope.sender) else @buffer << envelope distribute Behaviour::MESSAGE_PROCESSED end end |