Add Pool

Author: pitr-ch

lib/concurrent/actor/utils.rb

@@ -1,8 +1,10 @@
 module Concurrent
   module Actor
     module Utils
-      require 'concurrent/actor/utils/broadcast'
       require 'concurrent/actor/utils/ad_hoc'
+      require 'concurrent/actor/utils/broadcast'
+      require 'concurrent/actor/utils/balancer'
+      require 'concurrent/actor/utils/pool'
     end
   end
 end

lib/concurrent/actor/utils/balancer.rb

@@ -0,0 +1,40 @@
+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.
+      # @see Pool
+      class Balancer < RestartingContext
+
+        def initialize
+          @receivers = []
+          @buffer    = []
+        end
+
+        def on_message(message)
+          case message
+          when :subscribe
+            @receivers << envelope.sender
+            distribute
+            true
+          when :unsubscribe
+            @receivers.delete envelope.sender
+            true
+          when :subscribed?
+            @receivers.include? envelope.sender
+          else
+            @buffer << message
+            distribute
+          end
+        end
+
+        def distribute
+          while !@receivers.empty? && !@buffer.empty?
+            @receivers.shift << @buffer.shift
+          end
+        end
+      end
+    end
+  end
+end

lib/concurrent/actor/utils/broadcast.rb

@@ -7,7 +7,7 @@ module Utils
       # Allows to build pub/sub easily.
       # @example news
       #   news_channel = Concurrent::Actor::Utils::Broadcast.spawn :news
-
+      #
       #   2.times do |i|
       #     Concurrent::Actor::Utils::AdHoc.spawn "listener-#{i}" do
       #       news_channel << :subscribe
@@ -17,7 +17,7 @@ module Utils
       #
       #   news_channel << 'Ruby rocks!'
       #   # prints: 'Ruby rocks!' twice
-      class Broadcast < Context
+      class Broadcast < RestartingContext
 
         def initialize
           @receivers = Set.new

lib/concurrent/actor/utils/pool.rb

@@ -0,0 +1,59 @@
+require 'concurrent/actor/utils/balancer'
+
+module Concurrent
+  module Actor
+    module Utils
+
+      # Allows to create a pool of workers and distribute work between them
+      # @param [Integer] size number of workers
+      # @yield [balancer, index] a block spawning an worker instance. called +size+ times.
+      #   The worker should be descendant of AbstractWorker and supervised, see example.
+      # @yieldparam [Balancer] balancer to pass to the worker
+      # @yieldparam [Integer] index of the worker, usually used in its name
+      # @yieldreturn [Reference] the reference of newly created worker
+      # @example
+      #     class Worker < Concurrent::Actor::Utils::AbstractWorker
+      #       def work(message)
+      #         p message * 5
+      #       end
+      #     end
+      #
+      #     pool = Concurrent::Actor::Utils::Pool.spawn! 'pool', 5 do |balancer, index|
+      #       Worker.spawn name: "worker-#{index}", supervise: true, args: [balancer]
+      #     end
+      #
+      #     pool << 'asd' << 2
+      #     # prints:
+      #     # "asdasdasdasdasd"
+      #     # 10
+      class Pool < RestartingContext
+        def initialize(size, &worker_initializer)
+          @balancer = Balancer.spawn name: :balancer, supervise: true
+          @workers  = Array.new(size, &worker_initializer.curry[@balancer])
+          @workers.each { |w| Type! w, Reference }
+        end
+
+        def on_message(message)
+          @balancer << message
+        end
+      end
+
+      class AbstractWorker < RestartingContext
+        def initialize(balancer)
+          @balancer = balancer
+          @balancer << :subscribe
+        end
+
+        def on_message(message)
+          work message
+        ensure
+          @balancer << :subscribe
+        end
+
+        def work(message)
+          raise NotImplementedError
+        end
+      end
+    end
+  end
+end