Merge pull request #321 from ruby-concurrency/internal-reorg-channel

Moved all channel classes into the Channel module.

Author: jdantonio

lib/concurrent/channel/blocking_ring_buffer.rb

@@ -1,77 +1,80 @@
 require 'concurrent/synchronization'
 
 module Concurrent
-  class BlockingRingBuffer < Synchronization::Object
+  module Channel
 
-    def initialize(capacity)
-      super()
-      synchronize { ns_initialize capacity}
-    end
+    class BlockingRingBuffer < Synchronization::Object
 
-    # @return [Integer] the capacity of the buffer
-    def capacity
-      synchronize { @buffer.capacity }
-    end
+      def initialize(capacity)
+        super()
+        synchronize { ns_initialize capacity}
+      end
 
-    # @return [Integer] the number of elements currently in the buffer
-    def count
-      synchronize { @buffer.count }
-    end
+      # @return [Integer] the capacity of the buffer
+      def capacity
+        synchronize { @buffer.capacity }
+      end
 
-    # @return [Boolean] true if buffer is empty, false otherwise
-    def empty?
-      synchronize { @buffer.empty? }
-    end
+      # @return [Integer] the number of elements currently in the buffer
+      def count
+        synchronize { @buffer.count }
+      end
 
-    # @return [Boolean] true if buffer is full, false otherwise
-    def full?
-      synchronize { @buffer.full? }
-    end
+      # @return [Boolean] true if buffer is empty, false otherwise
+      def empty?
+        synchronize { @buffer.empty? }
+      end
 
-    # @param [Object] value the value to be inserted
-    # @return [Boolean] true if value has been inserted, false otherwise
-    def put(value)
-      synchronize do
-        wait_while_full
-        @buffer.offer(value)
-        ns_signal
-        true
+      # @return [Boolean] true if buffer is full, false otherwise
+      def full?
+        synchronize { @buffer.full? }
       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
-      synchronize do
-        wait_while_empty
-        result = @buffer.poll
-        ns_signal
-        result
+      # @param [Object] value the value to be inserted
+      # @return [Boolean] true if value has been inserted, false otherwise
+      def put(value)
+        synchronize do
+          wait_while_full
+          @buffer.offer(value)
+          ns_signal
+          true
+        end
       end
-    end
 
-    # @return [Object] the first available value and without removing it from
-    #   the buffer. If buffer is empty returns nil
-    def peek
-      synchronize { @buffer.peek }
-    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
+        synchronize do
+          wait_while_empty
+          result = @buffer.poll
+          ns_signal
+          result
+        end
+      end
 
-    protected
+      # @return [Object] the first available value and without removing it from
+      #   the buffer. If buffer is empty returns nil
+      def peek
+        synchronize { @buffer.peek }
+      end
 
-    def ns_initialize(capacity)
-      @buffer = RingBuffer.new(capacity)
-      @first = @last = 0
-      @count = 0
-    end
+      protected
 
-    private
+      def ns_initialize(capacity)
+        @buffer = RingBuffer.new(capacity)
+        @first = @last = 0
+        @count = 0
+      end
 
-    def wait_while_full
-      ns_wait_until { !@buffer.full? }
-    end
+      private
 
-    def wait_while_empty
-      ns_wait_until { !@buffer.empty? }
+      def wait_while_full
+        ns_wait_until { !@buffer.full? }
+      end
+
+      def wait_while_empty
+        ns_wait_until { !@buffer.empty? }
+      end
     end
   end
 end

lib/concurrent/channel/buffered_channel.rb

@@ -1,82 +1,85 @@
 require 'concurrent/channel/waitable_list'
 
 module Concurrent
-  class BufferedChannel
+  module Channel
 
-    def initialize(size)
-      @mutex = Mutex.new
-      @buffer_condition = ConditionVariable.new
+    class BufferedChannel
 
-      @probe_set = WaitableList.new
-      @buffer = RingBuffer.new(size)
-    end
+      def initialize(size)
+        @mutex = Mutex.new
+        @buffer_condition = ConditionVariable.new
 
-    def probe_set_size
-      @probe_set.size
-    end
+        @probe_set = WaitableList.new
+        @buffer = RingBuffer.new(size)
+      end
 
-    def buffer_queue_size
-      @mutex.synchronize { @buffer.count }
-    end
+      def probe_set_size
+        @probe_set.size
+      end
 
-    def push(value)
-      until set_probe_or_push_into_buffer(value)
+      def buffer_queue_size
+        @mutex.synchronize { @buffer.count }
       end
-    end
 
-    def pop
-      probe = Channel::Probe.new
-      select(probe)
-      probe.value
-    end
+      def push(value)
+        until set_probe_or_push_into_buffer(value)
+        end
+      end
 
-    def select(probe)
-      @mutex.synchronize do
+      def pop
+        probe = Channel::Probe.new
+        select(probe)
+        probe.value
+      end
 
-        if @buffer.empty?
-          @probe_set.put(probe)
-          true
-        else
-          shift_buffer if probe.try_set([peek_buffer, self])
-        end
+      def select(probe)
+        @mutex.synchronize do
+
+          if @buffer.empty?
+            @probe_set.put(probe)
+            true
+          else
+            shift_buffer if probe.try_set([peek_buffer, self])
+          end
 
+        end
       end
-    end
 
-    def remove_probe(probe)
-      @probe_set.delete(probe)
-    end
+      def remove_probe(probe)
+        @probe_set.delete(probe)
+      end
 
-    private
+      private
 
-    def push_into_buffer(value)
-      @buffer_condition.wait(@mutex) while @buffer.full?
-      @buffer.offer value
-      @buffer_condition.broadcast
-    end
+      def push_into_buffer(value)
+        @buffer_condition.wait(@mutex) while @buffer.full?
+        @buffer.offer value
+        @buffer_condition.broadcast
+      end
 
-    def peek_buffer
-      @buffer_condition.wait(@mutex) while @buffer.empty?
-      @buffer.peek
-    end
+      def peek_buffer
+        @buffer_condition.wait(@mutex) while @buffer.empty?
+        @buffer.peek
+      end
 
-    def shift_buffer
-      @buffer_condition.wait(@mutex) while @buffer.empty?
-      result = @buffer.poll
-      @buffer_condition.broadcast
-      result
-    end
+      def shift_buffer
+        @buffer_condition.wait(@mutex) while @buffer.empty?
+        result = @buffer.poll
+        @buffer_condition.broadcast
+        result
+      end
 
-    def set_probe_or_push_into_buffer(value)
-      @mutex.synchronize do
-        if @probe_set.empty?
-          push_into_buffer(value)
-          true
-        else
-          @probe_set.take.try_set([value, self])
+      def set_probe_or_push_into_buffer(value)
+        @mutex.synchronize do
+          if @probe_set.empty?
+            push_into_buffer(value)
+            true
+          else
+            @probe_set.take.try_set([value, self])
+          end
         end
       end
-    end
 
+    end
   end
 end

lib/concurrent/channel/ring_buffer.rb

@@ -1,60 +1,62 @@
 module Concurrent
+  module Channel
+
+    # non-thread safe buffer
+    class RingBuffer
+
+      def initialize(capacity)
+        @buffer = Array.new(capacity)
+        @first = @last = 0
+        @count = 0
+      end
+
+
+      # @return [Integer] the capacity of the buffer
+      def capacity
+        @buffer.size
+      end
+
+      # @return [Integer] the number of elements currently in the buffer
+      def count
+        @count
+      end
+
+      # @return [Boolean] true if buffer is empty, false otherwise
+      def empty?
+        @count == 0
+      end
+
+      # @return [Boolean] true if buffer is full, false otherwise
+      def full?
+        @count == capacity
+      end
+
+      # @param [Object] value
+      # @return [Boolean] true if value has been inserted, false otherwise
+      def offer(value)
+        return false if full?
+
+        @buffer[@last] = value
+        @last = (@last + 1) % @buffer.size
+        @count += 1
+        true
+      end
+
+      # @return [Object] the first available value and removes it from the buffer. If buffer is empty returns nil
+      def poll
+        result = @buffer[@first]
+        @buffer[@first] = nil
+        @first = (@first + 1) % @buffer.size
+        @count -= 1
+        result
+      end
+
+      # @return [Object] the first available value and without removing it from
+      #   the buffer. If buffer is empty returns nil
+      def peek
+        @buffer[@first]
+      end
 
-  # non-thread safe buffer
-  class RingBuffer
-
-    def initialize(capacity)
-      @buffer = Array.new(capacity)
-      @first = @last = 0
-      @count = 0
-    end
-
-
-    # @return [Integer] the capacity of the buffer
-    def capacity
-      @buffer.size
-    end
-
-    # @return [Integer] the number of elements currently in the buffer
-    def count
-      @count
-    end
-
-    # @return [Boolean] true if buffer is empty, false otherwise
-    def empty?
-      @count == 0
-    end
-
-    # @return [Boolean] true if buffer is full, false otherwise
-    def full?
-      @count == capacity
     end
-
-    # @param [Object] value
-    # @return [Boolean] true if value has been inserted, false otherwise
-    def offer(value)
-      return false if full?
-
-      @buffer[@last] = value
-      @last = (@last + 1) % @buffer.size
-      @count += 1
-      true
-    end
-
-    # @return [Object] the first available value and removes it from the buffer. If buffer is empty returns nil
-    def poll
-      result = @buffer[@first]
-      @buffer[@first] = nil
-      @first = (@first + 1) % @buffer.size
-      @count -= 1
-      result
-    end
-
-    # @return [Object] the first available value and without removing it from
-    #   the buffer. If buffer is empty returns nil
-    def peek
-      @buffer[@first]
-    end
-
   end
 end