Updated channels with Synchronization::Object.

Author: jdantonio

lib/concurrent/channel/blocking_ring_buffer.rb

@@ -1,73 +1,76 @@
-require 'concurrent/atomic/condition'
+require 'concurrent/synchronization'
 
 module Concurrent
-  class BlockingRingBuffer
+  class BlockingRingBuffer < Synchronization::Object
 
     def initialize(capacity)
-      @buffer = RingBuffer.new(capacity)
-      @first = @last = 0
-      @count = 0
-      @mutex = Mutex.new
-      @condition = Condition.new
+      super(capacity)
     end
 
     # @return [Integer] the capacity of the buffer
     def capacity
-      @mutex.synchronize { @buffer.capacity }
+      synchronize { @buffer.capacity }
     end
 
     # @return [Integer] the number of elements currently in the buffer
     def count
-      @mutex.synchronize { @buffer.count }
+      synchronize { @buffer.count }
     end
 
     # @return [Boolean] true if buffer is empty, false otherwise
     def empty?
-      @mutex.synchronize { @buffer.empty? }
+      synchronize { @buffer.empty? }
     end
 
     # @return [Boolean] true if buffer is full, false otherwise
     def full?
-      @mutex.synchronize { @buffer.full? }
+      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
+      synchronize do
         wait_while_full
         @buffer.offer(value)
-        @condition.signal
+        ns_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
+      synchronize do
         wait_while_empty
         result = @buffer.poll
-        @condition.signal
+        ns_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 }
+      synchronize { @buffer.peek }
+    end
+
+    protected
+
+    def ns_initialize(capacity)
+      @buffer = RingBuffer.new(capacity)
+      @first = @last = 0
+      @count = 0
     end
 
     private
 
     def wait_while_full
-      @condition.wait(@mutex) while @buffer.full?
+      ns_wait_until { !@buffer.full? }
     end
 
     def wait_while_empty
-      @condition.wait(@mutex) while @buffer.empty?
+      ns_wait_until { !@buffer.empty? }
     end
-
   end
 end

lib/concurrent/channel/buffered_channel.rb

@@ -1,6 +1,5 @@
 require 'concurrent/atomic/condition'
-
-require_relative 'waitable_list'
+require 'concurrent/channel/waitable_list'
 
 module Concurrent
   class BufferedChannel

lib/concurrent/channel/waitable_list.rb

@@ -1,40 +1,38 @@
-require 'concurrent/atomic/condition'
+require 'concurrent/synchronization'
 
 module Concurrent
-  class WaitableList
-
-    def initialize
-      @mutex = Mutex.new
-      @condition = Condition.new
-
-      @list = []
-    end
+  class WaitableList < Synchronization::Object
 
     def size
-      @mutex.synchronize { @list.size }
+      synchronize { @list.size }
     end
 
     def empty?
-      @mutex.synchronize { @list.empty? }
+      synchronize { @list.empty? }
     end
 
     def put(value)
-      @mutex.synchronize do
+      synchronize do
         @list << value
-        @condition.signal
+        ns_signal
       end
     end
 
     def delete(value)
-      @mutex.synchronize { @list.delete(value) }
+      synchronize { @list.delete(value) }
     end
 
     def take
-      @mutex.synchronize do
-        @condition.wait(@mutex) while @list.empty?
+      synchronize do
+        ns_wait_until { !@list.empty? }
         @list.shift
       end
     end
 
+    protected
+
+    def ns_initialize
+      @list = []
+    end
   end
 end

lib/concurrent/synchronization/abstract_object.rb

@@ -58,7 +58,7 @@ def ns_initialize(*args, &block)
       #     synchronize { ns_wait_until(timeout, &condition) }
       #   end
       #   ```
-      def ns_wait_until(timeout, &condition)
+      def ns_wait_until(timeout = nil, &condition)
         if timeout
           wait_until = Concurrent.monotonic_time + timeout
           loop do