Fixed channel ticker/timer closing and next.

Author: jdantonio

lib/concurrent/agent.rb

@@ -166,6 +166,7 @@ class Agent < Synchronization::LockableObject
     class Error < StandardError
       def initialize(message = nil)
         message ||= 'agent must be restarted before jobs can post'
+        super(message)
       end
     end
 
@@ -174,6 +175,7 @@ def initialize(message = nil)
     class ValidationError < Error
       def initialize(message = nil)
         message ||= 'invalid value'
+        super(message)
       end
     end
 

lib/concurrent/channel.rb

@@ -116,20 +116,25 @@ def offer?(item)
     end
 
     def take
-      item, _ = self.next
-      item
+      item = do_take
+      item == Buffer::NO_VALUE ? nil : item
     end
     alias_method :receive, :take
     alias_method :~, :take
 
     def take!
-      item, _ = do_next
+      item = do_take
       raise Error if item == Buffer::NO_VALUE
       item
     end
 
     def take?
-      item, _ = self.next?
+      item = do_take
+      item = if item == Buffer::NO_VALUE
+               Concurrent::Maybe.nothing
+             else
+               Concurrent::Maybe.just(item)
+             end
       item
     end
 
@@ -238,13 +243,21 @@ def go_loop_via(executor, *args, &block)
 
     private
 
-    def validator() @validator; end
+    def validator
+      @validator
+    end
 
-    def validator=(value) @validator = value; end
+    def validator=(value)
+      @validator = value
+    end
 
-    def buffer() @buffer; end
+    def buffer
+      @buffer
+    end
 
-    def buffer=(value) @buffer = value; end
+    def buffer=(value)
+      @buffer = value
+    end
 
     def validate(value, allow_nil, raise_error)
       if !allow_nil && value.nil?
@@ -267,6 +280,10 @@ def do_offer(item)
       buffer.offer(item)
     end
 
+    def do_take
+      buffer.take
+    end
+
     def do_next
       buffer.next
     end

lib/concurrent/channel/buffer/base.rb

@@ -193,15 +193,25 @@ def closed?
 
         private
 
-        def buffer() @buffer; end
+        def buffer
+          @buffer
+        end
 
-        def buffer=(value) @buffer = value; end
+        def buffer=(value)
+          @buffer = value
+        end
 
-        def closed=(value) @closed = value; end
+        def closed=(value)
+          @closed = value
+        end
 
-        def capacity=(value) @capacity = value; end
+        def capacity=(value)
+          @capacity = value
+        end
 
-        def size=(value) @size = value; end
+        def size=(value)
+          @size = value
+        end
 
         def ns_initialize(*args)
         end

lib/concurrent/channel/buffer/ticker.rb

@@ -17,34 +17,26 @@ def offer(item)
         end
 
         def take
+          # a Go timer will block forever if stopped
           loop do
-            result, _ = do_poll
-            if result.nil?
-              return NO_VALUE
-            elsif result != NO_VALUE
-              return result
-            end
+            tick = do_poll
+            return tick if tick != NO_VALUE
+            Thread.pass
           end
         end
 
         def next
+          # a Go timer will block forever if stopped
+          # it will always return `true` for more
           loop do
-            result, _ = do_poll
-            if result.nil?
-              return NO_VALUE, false
-            elsif result != NO_VALUE
-              return result, true
-            end
+            tick = do_poll
+            return tick, true if tick != NO_VALUE
+            Thread.pass
           end
         end
 
         def poll
-          result, _ = do_poll
-          if result.nil? || result == NO_VALUE
-            NO_VALUE
-          else
-            result
-          end
+          do_poll
         end
 
         private
@@ -55,21 +47,27 @@ def ns_initialize(interval)
           self.capacity = 1
         end
 
-        def ns_size() 0; end
+        def ns_size
+          0
+        end
 
-        def ns_empty?() false; end
+        def ns_empty?
+          false
+        end
 
-        def ns_full?() true; end
+        def ns_full?
+          true
+        end
 
         def do_poll
-          if ns_closed?
-            return nil, false
-          elsif (now = Concurrent.monotonic_time) > @next_tick
-            tick = Concurrent::Channel::Tick.new(@next_tick)
-            @next_tick = now + @interval
-            return tick, true
-          else
-            return NO_VALUE, true
+          synchronize do
+            if !ns_closed? && (now = Concurrent.monotonic_time) >= @next_tick
+              tick = Concurrent::Channel::Tick.new(@next_tick)
+              @next_tick = now + @interval
+              return tick
+            else
+              return NO_VALUE
+            end
           end
         end
       end

lib/concurrent/channel/buffer/timer.rb

@@ -17,35 +17,26 @@ def offer(item)
         end
 
         def take
+          # a Go timer will block forever if stopped
           loop do
-            result, tick = do_poll
-            if result == :closed
-              return NO_VALUE
-            elsif result == :tick
-              return tick
-            end
+            tick = do_poll
+            return tick if tick != NO_VALUE
             Thread.pass
           end
         end
 
         def next
+          # a Go timer will block forever if stopped
+          # it will always return `true` for more
           loop do
-            status, tick = do_poll
-            if status == :closed
-              return NO_VALUE, false
-            elsif status == :tick
-              return tick, false
-              # AFAIK a Go timer will block forever if stopped
-              #elsif status == :closed
-              #return false, false
-            end
+            tick = do_poll
+            return tick, true if tick != NO_VALUE
             Thread.pass
           end
         end
 
         def poll
-          status, tick = do_poll
-          status == :tick ? tick : NO_VALUE
+          do_poll
         end
 
         private
@@ -55,22 +46,26 @@ def ns_initialize(delay)
           self.capacity = 1
         end
 
-        def ns_size() 0; end
+        def ns_size
+          0
+        end
 
-        def ns_empty?() false; end
+        def ns_empty?
+          false
+        end
 
-        def ns_full?() true; end
+        def ns_full?
+          true
+        end
 
         def do_poll
           synchronize do
-            if ns_closed?
-              return :closed, false
-            elsif Concurrent.monotonic_time > @tick
+            if !ns_closed? && Concurrent.monotonic_time >= @tick
               # only one listener gets notified
               self.closed = true
-              return :tick, Concurrent::Channel::Tick.new(@tick)
+              return Concurrent::Channel::Tick.new(@tick)
             else
-              return :wait, true
+              return NO_VALUE
             end
           end
         end

spec/concurrent/channel/buffer/ticker_spec.rb

@@ -4,13 +4,13 @@ module Concurrent::Channel::Buffer
 
   describe Ticker do
 
-    subject { described_class.new(1) }
+    let(:delay) { 0.1 }
+    subject { described_class.new(delay) }
 
     it_behaves_like :channel_timing_buffer
 
     context '#take' do
       it 'triggers until closed' do
-        subject = described_class.new(0.1)
         expected = 3
         actual = 0
         expected.times { actual += 1 if subject.take.is_a? Concurrent::Channel::Tick }
@@ -20,7 +20,6 @@ module Concurrent::Channel::Buffer
 
     context '#poll' do
       it 'triggers until closed' do
-        subject = described_class.new(0.1)
         expected = 3
         actual = 0
         expected.times do
@@ -32,15 +31,7 @@ module Concurrent::Channel::Buffer
     end
 
     context '#next' do
-
-      it 'returns more until closed' do
-        subject = described_class.new(0.1)
-        _, more = subject.next
-        expect(more).to be true
-      end
-
       it 'triggers until closed' do
-        subject = described_class.new(0.1)
         expected = 3
         actual = 0
         expected.times { actual += 1 if subject.next.first.is_a? Concurrent::Channel::Tick }

spec/concurrent/channel/buffer/timer_spec.rb

@@ -4,21 +4,20 @@ module Concurrent::Channel::Buffer
 
   describe Timer do
 
-    subject { described_class.new(0) }
+    let(:delay) { 0.1 }
+    subject { described_class.new(0.1) }
 
     it_behaves_like :channel_timing_buffer
 
     context '#take' do
       it 'closes automatically on first take' do
-        subject = described_class.new(0.1)
         expect(subject.take).to be_truthy
         expect(subject).to be_closed
       end
     end
 
     context '#poll' do
       it 'closes automatically on first take' do
-        subject = described_class.new(0.1)
         loop do
           break if subject.poll != NO_VALUE
         end
@@ -27,25 +26,13 @@ module Concurrent::Channel::Buffer
     end
 
     context '#next' do
-
       it 'closes automatically on first take' do
-        subject = described_class.new(0.1)
         loop do
           value, _ = subject.next
           break if value != NO_VALUE
         end
         expect(subject).to be_closed
       end
-
-      it 'returns false for more on first take' do
-        subject = described_class.new(0.1)
-        more = true
-        loop do
-          value, more = subject.next
-          break if value != NO_VALUE
-        end
-        expect(more).to be false
-      end
     end
   end
 end