initial unbuffered channel implementation

Author: mighe

lib/concurrent.rb

@@ -32,6 +32,8 @@
 require 'concurrent/tvar'
 require 'concurrent/utilities'
 
+require 'concurrent/channel/unbuffered_channel'
+
 require 'concurrent/cached_thread_pool'
 require 'concurrent/fixed_thread_pool'
 require 'concurrent/immediate_executor'

lib/concurrent/channel/unbuffered_channel.rb

@@ -0,0 +1,38 @@
+module Concurrent
+  class UnbufferedChannel
+
+    def initialize
+      @mutex = Mutex.new
+      @condition = Condition.new
+
+      @wait_set = []
+    end
+
+    def push(value)
+      probe = @mutex.synchronize do
+        @condition.wait(@mutex) while @wait_set.empty?
+        @wait_set.shift
+      end
+
+      probe.set(value)
+    end
+
+    def pop
+      probe = IVar.new
+
+      @mutex.synchronize do
+        @wait_set << probe
+        @condition.signal
+      end
+
+      probe.value
+    end
+
+    def select(probe)
+    end
+
+    def remove_probe(probe)
+    end
+
+  end
+end

spec/concurrent/channel/unbuffered_channel_spec.rb

@@ -0,0 +1,47 @@
+require 'spec_helper'
+
+module Concurrent
+
+  describe UnbufferedChannel do
+
+    let!(:channel) { subject } # let is not thread safe, let! creates the object before ensuring uniqueness
+
+    context 'with one thread' do
+
+      context 'without timeout' do
+
+        describe '#push' do
+          it 'should block' do
+            t = Thread.new { channel.push 5 }
+            sleep(0.05)
+            t.status.should eq 'sleep'
+          end
+        end
+
+        describe '#pop' do
+          it 'should block' do
+            t = Thread.new { channel.pop }
+            sleep(0.05)
+            t.status.should eq 'sleep'
+          end
+        end
+
+      end
+
+    end
+
+    context 'cooperating threads' do
+      it 'passes the pushed value to thread waiting on pop' do
+        result = nil
+
+        Thread.new { channel.push 42 }
+        Thread.new { result = channel.pop }
+
+        sleep(0.05)
+
+        result.should eq 42
+      end
+    end
+
+  end
+end