Contents

require 'concurrent/actor/context'

module Concurrent
  module Actor
    module Utils

      # 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 Pool
      class Balancer < RestartingContext

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

        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

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

Version data entries

3 entries across 3 versions & 1 rubygems

Version	Path
concurrent-ruby-edge-0.7.2	lib/concurrent-ruby-edge/concurrent/actor/utils/balancer.rb
concurrent-ruby-edge-0.7.1	lib/concurrent-ruby-edge/concurrent/actor/utils/balancer.rb
concurrent-ruby-edge-0.7.0	lib/concurrent-ruby-edge/concurrent/actor/utils/balancer.rb