Channel::Probe is now just an IVar.

Author: jdantonio

lib/concurrent/channel/buffered_channel.rb

@@ -40,7 +40,7 @@ def select(probe)
           @probe_set.put(probe)
           true
         else
-          shift_buffer if probe.set_unless_assigned(peek_buffer, self)
+          shift_buffer if probe.set?([peek_buffer, self])
         end
 
       end
@@ -76,7 +76,7 @@ def set_probe_or_push_into_buffer(value)
           push_into_buffer(value)
           true
         else
-          @probe_set.take.set_unless_assigned(value, self)
+          @probe_set.take.set?([value, self])
         end
       end
     end

lib/concurrent/channel/channel.rb

@@ -3,37 +3,12 @@
 module Concurrent
   module Channel
 
-    class Probe < Concurrent::IVar
-
-      def initialize(value = NO_VALUE, opts = {})
-        super(value, opts)
-      end
-
-      def set_unless_assigned(value, channel)
-        mutex.synchronize do
-          return false if [:fulfilled, :rejected].include? @state
-
-          set_state(true, [value, channel], nil)
-          event.set
-          true
-        end
-      end
-
-      alias_method :composite_value, :value
-
-      def value
-        composite_value.nil? ? nil : composite_value[0]
-      end
-
-      def channel
-        composite_value.nil? ? nil : composite_value[1]
-      end
-    end
+    Probe = IVar
 
     def self.select(*channels)
       probe = Probe.new
       channels.each { |channel| channel.select(probe) }
-      result = probe.composite_value
+      result = probe.value
       channels.each { |channel| channel.remove_probe(probe) }
       result
     end

lib/concurrent/channel/unbuffered_channel.rb

@@ -12,8 +12,7 @@ def probe_set_size
     end
 
     def push(value)
-      # TODO set_unless_assigned define on IVar as #set_state? or #try_set_state
-      until @probe_set.take.set_unless_assigned(value, self)
+      until @probe_set.take.set?([value, self])
       end
     end
 

lib/concurrent/future.rb

@@ -88,8 +88,6 @@ def set(value = IVar::NO_VALUE, &block)
       execute
     end
 
-    protected :complete
-
     private
 
     # @!visibility private

lib/concurrent/ivar.rb

@@ -103,11 +103,8 @@ def add_observer(observer = nil, func = :update, &block)
     #   been set or otherwise completed
     # @return [IVar] self
     def set(value = NO_VALUE)
-      if (block_given? && value != NO_VALUE) || (!block_given? && value == NO_VALUE)
-        raise ArgumentError.new('must set with either a value or a block')
-      elsif ! compare_and_set_state(:processing, :pending)
-        raise MultipleAssignmentError
-      end
+      check_for_block_or_value!(block_given?, value)
+      raise MultipleAssignmentError unless compare_and_set_state(:processing, :pending)
 
       begin
         value = yield if block_given?

spec/concurrent/channel/buffered_channel_spec.rb

@@ -47,7 +47,7 @@ module Concurrent
         it 'should assign value to a probe if probe set is not empty' do
           channel.select(probe)
           Thread.new { sleep(0.1); channel.push 3 }
-          expect(probe.value).to eq 3
+          expect(probe.value.first).to eq 3
         end
       end
 
@@ -62,14 +62,14 @@ module Concurrent
           channel.push 1
           result = channel.pop
 
-          expect(result).to eq 1
+          expect(result.first).to eq 1
         end
 
         it 'removes the first value from the buffer' do
           channel.push 'a'
           channel.push 'b'
 
-          expect(channel.pop).to eq 'a'
+          expect(channel.pop.first).to eq 'a'
           expect(channel.buffer_queue_size).to eq 1
         end
       end
@@ -91,7 +91,7 @@ module Concurrent
 
         Thread.new { channel.push 82 }
 
-        expect(probe.value).to eq 82
+        expect(probe.value.first).to eq 82
       end
 
     end
@@ -120,7 +120,7 @@ module Concurrent
 
           expect(channel.buffer_queue_size).to eq 1
 
-          expect(channel.pop).to eq 82
+          expect(channel.pop.first).to eq 82
 
         end
       end

spec/concurrent/channel/probe_spec.rb

@@ -20,57 +20,52 @@ def trigger_observable(observable)
       it_should_behave_like :observable
     end
 
-    describe '#set_unless_assigned' do
+    describe '#set?' do
       context 'empty probe' do
         it 'assigns the value' do
-          probe.set_unless_assigned(32, channel)
-          expect(probe.value).to eq 32
+          probe.set?([32, channel])
+          expect(probe.value.first).to eq 32
         end
 
         it 'assign the channel' do
-          probe.set_unless_assigned(32, channel)
-          expect(probe.channel).to be channel
+          probe.set?([32, channel])
+          expect(probe.value.last).to be channel
         end
 
         it 'returns true' do
-          expect(probe.set_unless_assigned('hi', channel)).to eq true
+          expect(probe.set?(['hi', channel])).to eq true
         end
       end
 
       context 'fulfilled probe' do
         before(:each) { probe.set([27, nil]) }
 
         it 'does not assign the value' do
-          probe.set_unless_assigned(88, channel)
-          expect(probe.value).to eq 27
+          probe.set?([88, channel])
+          expect(probe.value.first).to eq 27
         end
 
         it 'returns false' do
-          expect(probe.set_unless_assigned('hello', channel)).to eq false
+          expect(probe.set?(['hello', channel])).to eq false
         end
       end
 
       context 'rejected probe' do
         before(:each) { probe.fail }
 
         it 'does not assign the value' do
-          probe.set_unless_assigned(88, channel)
+          probe.set?([88, channel])
           expect(probe).to be_rejected
         end
 
         it 'has a nil value' do
           expect(probe.value).to be_nil
         end
 
-        it 'has a nil channel' do
-          expect(probe.channel).to be_nil
-        end
-
         it 'returns false' do
-          expect(probe.set_unless_assigned('hello', channel)).to eq false
+          expect(probe.set?(['hello', channel])).to eq false
         end
       end
     end
-
   end
 end

spec/concurrent/channel/unbuffered_channel_spec.rb

@@ -39,7 +39,7 @@ module Concurrent
 
         sleep(0.1)
 
-        expect(result).to eq 42
+        expect(result.first).to eq 42
       end
 
       it 'passes the pushed value to only one thread' do
@@ -62,7 +62,7 @@ module Concurrent
 
         sleep(0.1)
 
-        expect(result).to eq 57
+        expect(result.first).to eq 57
       end
     end
 
@@ -81,7 +81,7 @@ module Concurrent
 
         Thread.new { channel.push 82 }
 
-        expect(probe.value).to eq 82
+        expect(probe.value.first).to eq 82
       end
 
       it 'ignores already set probes and waits for a new one' do
@@ -101,7 +101,7 @@ module Concurrent
 
         sleep(0.05)
 
-        expect(new_probe.value).to eq 72
+        expect(new_probe.value.first).to eq 72
       end
 
     end