Merge pull request #131 from mastfish/patch-1

Add dataflow! to match value!

Author: jdantonio

lib/concurrent/dataflow.rb

@@ -66,12 +66,29 @@ def dataflow(*inputs, &block)
   module_function :dataflow
 
   def dataflow_with(executor, *inputs, &block)
+    call_dataflow(:value, executor, *inputs, &block)
+  end
+  module_function :dataflow_with
+  
+  def dataflow!(*inputs, &block)
+    dataflow_with!(Concurrent.configuration.global_task_pool, *inputs, &block)
+  end
+  module_function :dataflow!
+
+  def dataflow_with!(executor, *inputs, &block)
+    call_dataflow(:value!, executor, *inputs, &block)
+  end
+  module_function :dataflow_with!
+
+  private 
+
+  def call_dataflow(method, executor, *inputs, &block)
     raise ArgumentError.new('an executor must be provided') if executor.nil?
     raise ArgumentError.new('no block given') unless block_given?
     raise ArgumentError.new('not all dependencies are IVars') unless inputs.all? { |input| input.is_a? IVar }
 
     result = Future.new(executor: executor) do
-      values = inputs.map { |input| input.value }
+      values = inputs.map { |input| input.send(method) }
       block.call(*values)
     end
 
@@ -87,5 +104,5 @@ def dataflow_with(executor, *inputs, &block)
 
     result
   end
-  module_function :dataflow_with
+  module_function :call_dataflow
 end

spec/concurrent/dataflow_spec.rb

@@ -73,6 +73,19 @@ module Concurrent
       expect { Concurrent::dataflow_with(root_executor, nil, Future.execute{0}) }.to raise_error(ArgumentError)
     end
 
+    it 'doesn\'t raises exceptions from dependencies, unless called with !' do
+
+      d1 = Concurrent::dataflow(){raise}
+      d2 = Concurrent::dataflow(){raise}
+      f = Concurrent::dataflow!(d1, d2){|d1v, d2v| [d1v,d2v]}
+      expect{f.value!}.to raise_error
+
+      d1 = Concurrent::dataflow(){raise}
+      d2 = Concurrent::dataflow(){raise}
+      f = Concurrent::dataflow(d1, d2){|d1v, d2v| [d1v,d2v]}
+      expect{f.value!}.to_not raise_error
+    end
+
     it 'returns a Future' do
       expect(Concurrent::dataflow{0}).to be_a(Future)
       expect(Concurrent::dataflow{0}).to be_a(Future)
@@ -220,23 +233,7 @@ def fib_with_dot(n)
         sleep(0.1)
         expect(expected.value).to eq 377
       end
-
-      it 'can be called as Concurrent::dataflow and Concurrent::dataflow_with' do
-
-        def fib_with_colons(n)
-          if n < 2
-            Concurrent::dataflow { n }
-          else
-            n1 = fib_with_colons(n - 1)
-            n2 = fib_with_colons(n - 2)
-            Concurrent::dataflow_with(root_executor, n1, n2) { n1.value + n2.value }
-          end
-        end
-
-        expected = fib_with_colons(14)
-        sleep(0.1)
-        expect(expected.value).to eq 377
-      end
+      
     end
   end
 end