BufferedChannel is now a Synchronization::Object.

Author: jdantonio

lib/concurrent/channel/buffered_channel.rb

@@ -1,30 +1,28 @@
+require 'concurrent/synchronization'
 require 'concurrent/channel/waitable_list'
 
 module Concurrent
   module Channel
 
     # @api Channel
     # @!macro edge_warning
-    class BufferedChannel
+    class BufferedChannel < Synchronization::Object
 
       def initialize(size)
-        @mutex = Mutex.new
-        @buffer_condition = ConditionVariable.new
-
-        @probe_set = WaitableList.new
-        @buffer = RingBuffer.new(size)
+        super()
+        synchronize { ns_initialize(size) }
       end
 
       def probe_set_size
         @probe_set.size
       end
 
       def buffer_queue_size
-        @mutex.synchronize { @buffer.count }
+        synchronize { @buffer.count }
       end
 
       def push(value)
-        until set_probe_or_push_into_buffer(value)
+        until set_probe_or_ns_push_into_buffer(value)
         end
       end
 
@@ -35,53 +33,57 @@ def pop
       end
 
       def select(probe)
-        @mutex.synchronize do
-
+        synchronize do
           if @buffer.empty?
             @probe_set.put(probe)
             true
           else
-            shift_buffer if probe.try_set([peek_buffer, self])
+            ns_shift_buffer if probe.try_set([ns_peek_buffer, self])
           end
-
         end
       end
 
       def remove_probe(probe)
         @probe_set.delete(probe)
       end
 
+      protected
+
+      def ns_initialize(size)
+        @probe_set = WaitableList.new
+        @buffer = RingBuffer.new(size)
+      end
+
       private
 
-      def push_into_buffer(value)
-        @buffer_condition.wait(@mutex) while @buffer.full?
+      def ns_push_into_buffer(value)
+        ns_wait while @buffer.full?
         @buffer.offer value
-        @buffer_condition.broadcast
+        ns_broadcast
       end
 
-      def peek_buffer
-        @buffer_condition.wait(@mutex) while @buffer.empty?
+      def ns_peek_buffer
+        ns_wait while @buffer.empty?
         @buffer.peek
       end
 
-      def shift_buffer
-        @buffer_condition.wait(@mutex) while @buffer.empty?
+      def ns_shift_buffer
+        ns_wait while @buffer.empty?
         result = @buffer.poll
-        @buffer_condition.broadcast
+        ns_broadcast
         result
       end
 
-      def set_probe_or_push_into_buffer(value)
-        @mutex.synchronize do
+      def set_probe_or_ns_push_into_buffer(value)
+        synchronize do
           if @probe_set.empty?
-            push_into_buffer(value)
+            ns_push_into_buffer(value)
             true
           else
             @probe_set.take.try_set([value, self])
           end
         end
       end
-
     end
   end
 end