#2764

Committed 08 Dec 2014 03:40PM UTC coverage: 91.388% (-0.4%) from 91.753%

Build # #2764

Build Type

push

Committed by

Commit Message

Merge pull request #201 from rkday/fallback_handling

Posting to a shutdown thread pool - JRuby consistency and better naming

Run Details

18 of 26 new or added lines in 5 files covered. (69.23%)

212 existing lines in 36 files now uncovered.

2812 of 3077 relevant lines covered (91.39%)

369.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.0

/lib/concurrent/actor/utils/pool.rb

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)
          redirect @balancer
        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

1	require 'concurrent/actor/utils/balancer'	1✔
2
3	module Concurrent	1✔
4	module Actor	1✔
5	module Utils	1✔
6
7	# Allows to create a pool of workers and distribute work between them
8	# @param [Integer] size number of workers
9	# @yield [balancer, index] a block spawning an worker instance. called +size+ times.
10	# The worker should be descendant of AbstractWorker and supervised, see example.
11	# @yieldparam [Balancer] balancer to pass to the worker
12	# @yieldparam [Integer] index of the worker, usually used in its name
13	# @yieldreturn [Reference] the reference of newly created worker
14	# @example
15	# class Worker < Concurrent::Actor::Utils::AbstractWorker
16	# def work(message)
17	# p message * 5
18	# end
19	# end
20	#
21	# pool = Concurrent::Actor::Utils::Pool.spawn! 'pool', 5 do \|balancer, index\|
22	# Worker.spawn name: "worker-#{index}", supervise: true, args: [balancer]
23	# end
24	#
25	# pool << 'asd' << 2
26	# # prints:
27	# # "asdasdasdasdasd"
28	# # 10
29	class Pool < RestartingContext	1✔
30	def initialize(size, &worker_initializer)	1✔
31	@balancer = Balancer.spawn name: :balancer, supervise: true	1✔
32	@workers = Array.new(size, &worker_initializer.curry[@balancer])	1✔
33	@workers.each { \|w\| Type! w, Reference }	6✔
34	end
35
36	def on_message(message)	1✔
37	redirect @balancer	1✔
38	end
39	end
40
41	class AbstractWorker < RestartingContext	1✔
42	def initialize(balancer)	1✔
43	@balancer = balancer	5✔
44	@balancer << :subscribe	5✔
45	end
46
47	def on_message(message)	1✔
48	work message	1✔
49	ensure
50	@balancer << :subscribe	1✔
51	end
52
53	def work(message)	1✔
UNCOV 54	raise NotImplementedError	×
55	end
56	end
57	end
58	end
59	end

ruby-concurrency / concurrent-ruby / #2764

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous