123456789_123456789_123456789_123456789_123456789_

Class: Concurrent::Actor::Utils::Balancer

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.

See Also:

Constant Summary

Concern::Logging - Included

SEV_LABEL

Class Method Summary

Actor::AbstractContext - Inherited

.spawn

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

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

Instance Attribute Summary

Actor::AbstractContext - Inherited

#core

Actor::InternalDelegations - Included

#terminated?

Instance Method Summary

Actor::RestartingContext - Inherited

#behaviour_definition

Actor::AbstractContext - Inherited

#<<

Alias for AbstractContext#tell.
#ask,
#ask!

Alias for AbstractContext#ask.
#behaviour_definition,
#dead_letter_routing

Defines an actor responsible for dead letters.
#default_executor

override to se different default executor, e.g.
#default_reference_class

override if different class for reference is needed.
#envelope,
#on_event

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

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

tell self a message.
#initialize_core, #on_envelope

Actor::InternalDelegations - Included

#behaviour

see Core#behaviour
#behaviour!

see Core#behaviour!
#children, #context, #dead_letter_routing,
#log

delegates to core.log.
#redirect, #terminate!

Concern::Logging - Included

#log

Logs through Concurrent.global_logger, it can be overridden by setting @logger.

Actor::PublicDelegations - Included

#actor_class

Alias for PublicDelegations#context_class.
#context_class, #executor, #name, #parent, #path,
#ref

Alias for PublicDelegations#reference.
#reference

Actor::TypeCheck - Included

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

Constructor Details

.new ⇒ Balancer

[ GitHub ] 

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/utils/balancer.rb', line 13



def initialize
  @receivers = []
  @buffer    = []
end


  







Instance Method Details



  

    #distribute  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/utils/balancer.rb', line 37



def distribute
  while !@receivers.empty? && !@buffer.empty?
    redirect @receivers.shift, @buffer.shift
  end
end


  








  

    #on_message(message)  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby-edge/concurrent/actor/utils/balancer.rb', line 18



def on_message(message)
  command, who = message
  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