Merge pull request #432 from ruby-concurrency/channel-refactor

Channel refactor.

Author: jdantonio

lib/concurrent/channel.rb

@@ -8,8 +8,10 @@ module Concurrent
 
   # {include:file:doc/channel.md}
   class Channel
+    extend Forwardable
     include Enumerable
 
+    # NOTE: Move to global IO pool once stable
     GOROUTINES = Concurrent::CachedThreadPool.new
     private_constant :GOROUTINES
 
@@ -32,10 +34,17 @@ def initialize(message = nil)
       end
     end
 
+    def_delegators :buffer,
+      :size, :capacity, :close, :closed?,
+      :blocking?, :empty?, :full?
+
+    alias_method :length, :size
+    alias_method :stop, :close
+
     def initialize(opts = {})
       # undocumented -- for internal use only
       if opts.is_a? Buffer::Base
-        @buffer = opts
+        self.buffer = opts
         return
       end
 
@@ -45,25 +54,20 @@ def initialize(opts = {})
       if size && buffer == :unbuffered
         raise ArgumentError.new('unbuffered channels cannot have a size')
       elsif size.nil? && buffer.nil?
-        @buffer = BUFFER_TYPES[:unbuffered].new
+        self.buffer = BUFFER_TYPES[:unbuffered].new
       elsif size == 0 && buffer == :buffered
-        @buffer = BUFFER_TYPES[:unbuffered].new
+        self.buffer = BUFFER_TYPES[:unbuffered].new
       elsif buffer == :unbuffered
-        @buffer = BUFFER_TYPES[:unbuffered].new
+        self.buffer = BUFFER_TYPES[:unbuffered].new
       elsif size.nil? || size < 1
         raise ArgumentError.new('size must be at least 1 for this buffer type')
       else
         buffer ||= :buffered
-        @buffer = BUFFER_TYPES[buffer].new(size)
+        self.buffer = BUFFER_TYPES[buffer].new(size)
       end
 
-      @validator = opts.fetch(:validator, DEFAULT_VALIDATOR)
-    end
-
-    def size
-      @buffer.size
+      self.validator = opts.fetch(:validator, DEFAULT_VALIDATOR)
     end
-    alias_method :capacity, :size
 
     def put(item)
       return false unless validate(item, false, false)
@@ -129,22 +133,21 @@ def take?
       item
     end
 
+    # @example
     #
-    #   @example
+    #   jobs = Channel.new
     #
-    #     jobs = Channel.new
-    #
-    #     Channel.go do
-    #       loop do
-    #         j, more = jobs.next
-    #         if more
-    #           print "received job #{j}\n"
-    #         else
-    #           print "received all jobs\n"
-    #           break
-    #         end
+    #   Channel.go do
+    #     loop do
+    #       j, more = jobs.next
+    #       if more
+    #         print "received job #{j}\n"
+    #       else
+    #         print "received all jobs\n"
+    #         break
     #       end
     #     end
+    #   end
     def next
       item, more = do_next
       item = nil if item == Buffer::NO_VALUE
@@ -191,11 +194,6 @@ def each
       end
     end
 
-    def close
-      @buffer.close
-    end
-    alias_method :stop, :close
-
     class << self
       def timer(seconds)
         Channel.new(Buffer::Timer.new(seconds))
@@ -240,10 +238,12 @@ def go_loop_via(executor, *args, &block)
 
     private
 
+    attr_accessor :buffer, :validator
+
     def validate(value, allow_nil, raise_error)
       if !allow_nil && value.nil?
         raise_error ? raise(ValidationError.new('nil is not a valid value')) : false
-      elsif !@validator.call(value)
+      elsif !validator.call(value)
         raise_error ? raise(ValidationError) : false
       else
         true
@@ -254,19 +254,19 @@ def validate(value, allow_nil, raise_error)
     end
 
     def do_put(item)
-      @buffer.put(item)
+      buffer.put(item)
     end
 
     def do_offer(item)
-      @buffer.offer(item)
+      buffer.offer(item)
     end
 
     def do_next
-      @buffer.next
+      buffer.next
     end
 
     def do_poll
-      @buffer.poll
+      buffer.poll
     end
   end
 end

lib/concurrent/channel/buffer/base.rb

@@ -20,19 +20,26 @@ class Base < Synchronization::LockableObject
 
         # @!macro [attach] channel_buffer_size_reader
         #
+        #   The number of items currently in the buffer.
+        attr_reader :size
+
+        # @!macro [attach] channel_buffer_capacity_reader
+        #
         #   The maximum number of values which can be {#put} onto the buffer
         #   it becomes full.
-        attr_reader :size
-        alias_method :capacity, :size
+        attr_reader :capacity
 
         # @!macro [attach] channel_buffer_initialize
         #
         #   Creates a new buffer.
-        def initialize
+        def initialize(*args)
           super()
           synchronize do
             @closed = false
             @size = 0
+            @capacity = 0
+            @buffer = nil
+            ns_initialize(*args)
           end
         end
 
@@ -187,6 +194,12 @@ def closed?
 
         private
 
+        attr_accessor :buffer
+        attr_writer :closed, :capacity, :size
+
+        def ns_initialize(*args)
+        end
+
         # @!macro channel_buffer_closed_question
         def ns_closed?
           @closed

lib/concurrent/channel/buffer/buffered.rb

@@ -10,20 +10,6 @@ module Buffer
       # an item is removed from the buffer, creating spare capacity.
       class Buffered < Base
 
-        # @!macro channel_buffer_initialize
-        #
-        # @param [Integer] size the maximum capacity of the buffer; must be
-        #   greater than zero.
-        # @raise [ArgumentError] when the size is zero (0) or less.
-        def initialize(size)
-          raise ArgumentError.new('size must be greater than 0') if size.to_i <= 0
-          super()
-          synchronize do
-            @size = size.to_i
-            @buffer = []
-          end
-        end
-
         # @!macro channel_buffer_empty_question
         def empty?
           synchronize { ns_empty? }
@@ -85,7 +71,7 @@ def next
               if ns_closed? && ns_empty?
                 return NO_VALUE, false
               elsif !ns_empty?
-                item = @buffer.shift
+                item = buffer.shift
                 more = !ns_empty? || !ns_closed?
                 return item, more
               end
@@ -100,26 +86,37 @@ def poll
             if ns_empty?
               NO_VALUE
             else
-              @buffer.shift
+              buffer.shift
             end
           end
         end
 
         private
 
+        # @!macro channel_buffer_initialize
+        #
+        # @param [Integer] size the maximum capacity of the buffer; must be
+        #   greater than zero.
+        # @raise [ArgumentError] when the size is zero (0) or less.
+        def ns_initialize(size)
+          raise ArgumentError.new('size must be greater than 0') if size.to_i <= 0
+          self.capacity = size.to_i
+          self.buffer = []
+        end
+
         # @!macro channel_buffer_empty_question
         def ns_empty?
-          @buffer.length == 0
+          buffer.length == 0
         end
 
         # @!macro channel_buffer_full_question
         def ns_full?
-          @buffer.length == @size
+          buffer.length == capacity
         end
 
         # @!macro channel_buffer_put
         def ns_put_onto_buffer(item)
-          @buffer.push(item)
+          buffer.push(item)
         end
       end
     end

lib/concurrent/channel/buffer/dropping.rb

@@ -45,7 +45,7 @@ def ns_full?
 
         # @!macro channel_buffer_put
         def ns_put_onto_buffer(item)
-          @buffer.push(item) unless @buffer.size == size
+          buffer.push(item) unless buffer.size == capacity
         end
       end
     end

lib/concurrent/channel/buffer/sliding.rb

@@ -45,8 +45,8 @@ def ns_full?
 
         # @!macro channel_buffer_put
         def ns_put_onto_buffer(item)
-          @buffer.shift if @buffer.size == size
-          @buffer.push(item)
+          buffer.shift if buffer.size == capacity
+          buffer.push(item)
         end
       end
     end

lib/concurrent/channel/buffer/ticker.rb

@@ -8,14 +8,6 @@ module Buffer
 
       class Ticker < Base
 
-        def initialize(interval)
-          super()
-          synchronize do
-            @interval = interval.to_f
-            @next_tick = Concurrent.monotonic_time + interval
-          end
-        end
-
         def size() 1; end
 
         def empty?() false; end
@@ -63,6 +55,11 @@ def poll
 
         private
 
+        def ns_initialize(interval)
+          @interval = interval.to_f
+          @next_tick = Concurrent.monotonic_time + interval
+        end
+
         def do_poll
           if ns_closed?
             return nil, false

lib/concurrent/channel/buffer/timer.rb

@@ -8,15 +8,6 @@ module Buffer
 
       class Timer < Base
 
-        def initialize(delay)
-          super()
-          synchronize do
-            @tick = Concurrent.monotonic_time + delay.to_f
-            @closed = false
-            @empty = false
-          end
-        end
-
         def size() 1; end
 
         def empty?
@@ -59,13 +50,19 @@ def poll
 
         private
 
+        def ns_initialize(delay)
+          @tick = Concurrent.monotonic_time + delay.to_f
+          @closed = false
+          @empty = false
+        end
+
         def do_poll
           synchronize do
             return :closed, false if ns_closed?
 
             if Concurrent.monotonic_time > @tick
               # only one listener gets notified
-              @closed = @empty = true
+              closed = empty = true
               return :tick, Concurrent::Channel::Tick.new(@tick)
             else
               return :wait, true