#2701

Committed 12 Feb 2015 06:34PM UTC coverage: 92.31% (+0.6%) from 91.69%

Build # #2701

Build Type

push

Committed by

Commit Message

Merge pull request #238 from ruby-concurrency/semaphore-pure-java

Implementation of Concurrent::JavaSemaphore in pure Java.

Run Details

2881 of 3121 relevant lines covered (92.31%)

390.17 hits per line

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

85.71

/lib/concurrent/actor/utils/balancer.rb

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)
          case message
          when :subscribe
            @receivers << envelope.sender
            distribute
            true
          when :unsubscribe
            @receivers.delete envelope.sender
            true
          when :subscribed?
            @receivers.include? 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

1	module Concurrent	1✔
2	module Actor	1✔
3	module Utils	1✔
4
5	# Distributes messages between subscribed actors. Each actor'll get only one message then
6	# it's unsubscribed. The actor needs to resubscribe when it's ready to receive next message.
7	# It will buffer the messages if there is no worker registered.
8	# @see Pool
9	class Balancer < RestartingContext	1✔
10
11	def initialize	1✔
12	@receivers = []	1✔
13	@buffer = []	1✔
14	end
15
16	def on_message(message)	1✔
17	case message	7✔
18	when :subscribe
19	@receivers << envelope.sender	6✔
20	distribute	6✔
21	true	6✔
22	when :unsubscribe
23	@receivers.delete envelope.sender	×
24	true	×
25	when :subscribed?
26	@receivers.include? envelope.sender	×
27	else
28	@buffer << envelope	1✔
29	distribute	1✔
30	Behaviour::MESSAGE_PROCESSED	1✔
31	end
32	end
33
34	def distribute	1✔
35	while !@receivers.empty? && !@buffer.empty?	7✔
36	redirect @receivers.shift, @buffer.shift	1✔
37	end
38	end
39	end
40	end
41	end
42	end

ruby-concurrency / concurrent-ruby / #2701

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