implemented remove_probe

Author: mighe

lib/concurrent/channel/unbuffered_channel.rb

@@ -5,7 +5,11 @@ def initialize
       @mutex = Mutex.new
       @condition = Condition.new
 
-      @wait_set = []
+      @probe_set = []
+    end
+
+    def probe_set_size
+      @mutex.synchronize { @probe_set.size }
     end
 
     def push(value)
@@ -21,19 +25,20 @@ def pop
 
     def select(probe)
       @mutex.synchronize do
-        @wait_set << probe
+        @probe_set << probe
         @condition.signal
       end
     end
 
     def remove_probe(probe)
+      @mutex.synchronize { @probe_set.delete(probe) }
     end
 
     private
       def first_waiting_probe
         @mutex.synchronize do
-          @condition.wait(@mutex) while @wait_set.empty?
-          @wait_set.shift
+          @condition.wait(@mutex) while @probe_set.empty?
+          @probe_set.shift
         end
       end
 

spec/concurrent/channel/unbuffered_channel_spec.rb

@@ -4,7 +4,8 @@ module Concurrent
 
   describe UnbufferedChannel do
 
-    let!(:channel) { subject } # let is not thread safe, let! creates the object before ensuring uniqueness
+    let(:channel) { subject }
+    let(:probe) { Probe.new }
 
     context 'with one thread' do
 
@@ -58,8 +59,6 @@ module Concurrent
 
     describe 'select' do
 
-      let(:probe) { Probe.new }
-
       it 'does not block' do
         t = Thread.new { channel.select(probe) }
 
@@ -97,5 +96,26 @@ module Concurrent
       end
 
     end
+
+    describe 'probe set' do
+
+      it 'has size zero after creation' do
+        channel.probe_set_size.should eq 0
+      end
+
+      it 'increases size after a select' do
+        channel.select(probe)
+        channel.probe_set_size.should eq 1
+      end
+
+      it 'decreases size after a removal' do
+        channel.select(probe)
+        channel.remove_probe(probe)
+        channel.probe_set_size.should eq 0
+      end
+
+    end
+
+
   end
 end