#2872

Committed 08 Jun 2014 06:00PM UTC coverage: 77.573% (-11.7%) from 89.319%

Build # #2872

Build Type

push

Committed by

Commit Message

Updated badges with new repo location.

Coverage Stats

2186 of 2818 relevant lines covered (77.57%)

527.3 hits per line

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

38.89

/lib/concurrent/collection/blocking_ring_buffer.rb

require 'concurrent/atomic/condition'

module Concurrent
  class BlockingRingBuffer

    def initialize(capacity)
      @buffer = RingBuffer.new(capacity)
      @first = @last = 0
      @count = 0
      @mutex = Mutex.new
      @condition = Condition.new
    end

    # @return [Integer] the capacity of the buffer
    def capacity
      @mutex.synchronize { @buffer.capacity }
    end

    # @return [Integer] the number of elements currently in the buffer
    def count
      @mutex.synchronize { @buffer.count }
    end

    # @return [Boolean] true if buffer is empty, false otherwise
    def empty?
      @mutex.synchronize { @buffer.empty? }
    end

    # @return [Boolean] true if buffer is full, false otherwise
    def full?
      @mutex.synchronize { @buffer.full? }
    end

    # @param [Object] value the value to be inserted
    # @return [Boolean] true if value has been inserted, false otherwise
    def put(value)
      @mutex.synchronize do
        wait_while_full
        @buffer.offer(value)
        @condition.signal
        true
      end
    end

    # @return [Object] the first available value and removes it from the buffer. If buffer is empty it blocks until an element is available
    def take
      @mutex.synchronize do
        wait_while_empty
        result = @buffer.poll
        @condition.signal
        result
      end
    end

    # @return [Object] the first available value and without removing it from the buffer. If buffer is empty returns nil
    def peek
      @mutex.synchronize { @buffer.peek }
    end

    private

    def wait_while_full
      @condition.wait(@mutex) while @buffer.full?
    end

    def wait_while_empty
      @condition.wait(@mutex) while @buffer.empty?
    end

  end
end

1	require 'concurrent/atomic/condition'	1✔
2
3	module Concurrent	1✔
4	class BlockingRingBuffer	1✔
5
6	def initialize(capacity)	1✔
7	@buffer = RingBuffer.new(capacity)	×
8	@first = @last = 0	×
9	@count = 0	×
10	@mutex = Mutex.new	×
11	@condition = Condition.new	×
12	end
13
14	# @return [Integer] the capacity of the buffer
15	def capacity	1✔
16	@mutex.synchronize { @buffer.capacity }	×
17	end
18
19	# @return [Integer] the number of elements currently in the buffer
20	def count	1✔
21	@mutex.synchronize { @buffer.count }	×
22	end
23
24	# @return [Boolean] true if buffer is empty, false otherwise
25	def empty?	1✔
26	@mutex.synchronize { @buffer.empty? }	×
27	end
28
29	# @return [Boolean] true if buffer is full, false otherwise
30	def full?	1✔
31	@mutex.synchronize { @buffer.full? }	×
32	end
33
34	# @param [Object] value the value to be inserted
35	# @return [Boolean] true if value has been inserted, false otherwise
36	def put(value)	1✔
37	@mutex.synchronize do	×
38	wait_while_full	×
39	@buffer.offer(value)	×
40	@condition.signal	×
41	true	×
42	end
43	end
44
45	# @return [Object] the first available value and removes it from the buffer. If buffer is empty it blocks until an element is available
46	def take	1✔
47	@mutex.synchronize do	×
48	wait_while_empty	×
49	result = @buffer.poll	×
50	@condition.signal	×
51	result	×
52	end
53	end
54
55	# @return [Object] the first available value and without removing it from the buffer. If buffer is empty returns nil
56	def peek	1✔
57	@mutex.synchronize { @buffer.peek }	×
58	end
59
60	private	1✔
61
62	def wait_while_full	1✔
63	@condition.wait(@mutex) while @buffer.full?	×
64	end
65
66	def wait_while_empty	1✔
67	@condition.wait(@mutex) while @buffer.empty?	×
68	end
69
70	end
71	end

ruby-concurrency / concurrent-ruby / #2872

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