Removed Dataflow class and put #dataflow method directly on Concurrent module.

Author: jdantonio

README.md

@@ -107,12 +107,12 @@ These tools will help ease the burden, but at the end of the day it is essential
 ## Contributors
 
 * [Michele Della Torre](https://github.com/mighe)
+* [Chris Seaton](https://github.com/chrisseaton)
 * [Giuseppe Capizzi](https://github.com/gcapizzi)
 * [Brian Shirai](https://github.com/brixen)
 * [Chip Miller](https://github.com/chip-miller)
 * [Jamie Hodge](https://github.com/jamiehodge)
 * [Zander Hill](https://github.com/zph)
-* [Chris Seaton](https://github.com/chrisseaton)
 
 ## Contributing
 

lib/concurrent/dataflow.rb

@@ -2,64 +2,55 @@
 
 module Concurrent
 
-  module Dataflow
+  class AtomicCounter
 
-    class AtomicCounter
-
-      def initialize(init)
-        @counter = init
-        @mutex = Mutex.new
-      end
+    def initialize(init)
+      @counter = init
+      @mutex = Mutex.new
+    end
 
-      def decrement
-        @mutex.synchronize do
-          @counter -= 1
-        end
+    def decrement
+      @mutex.synchronize do
+        @counter -= 1
       end
-
     end
+  end
 
-    class DependencyCounter
+  class DependencyCounter
 
-      def initialize(count, &block)
-        @counter = AtomicCounter.new(count)
-        @block = block
-      end
+    def initialize(count, &block)
+      @counter = AtomicCounter.new(count)
+      @block = block
+    end
 
-      def update(time, value, reason)
-        if @counter.decrement == 0
-          @block.call()
-        end
+    def update(time, value, reason)
+      if @counter.decrement == 0
+        @block.call()
       end
-
     end
+  end
 
-    def self.dataflow(*inputs, &block)
-      raise ArgumentError.new('no block given') unless block_given?
-      raise ArgumentError.new('not all dependencies are Futures') unless inputs.all? { |input| input.is_a? Future }
+  def dataflow(*inputs, &block)
+    raise ArgumentError.new('no block given') unless block_given?
+    raise ArgumentError.new('not all dependencies are Futures') unless inputs.all? { |input| input.is_a? Future }
 
-      result = Concurrent::Future.new do
-        values = inputs.map { |input| input.value }
-        block.call(*values)
-      end
+    result = Future.new do
+      values = inputs.map { |input| input.value }
+      block.call(*values)
+    end
 
-      if inputs.empty?
-        result.execute
-      else
-        counter = Dataflow::DependencyCounter.new(inputs.size) { result.execute }
+    if inputs.empty?
+      result.execute
+    else
+      counter = DependencyCounter.new(inputs.size) { result.execute }
 
-        inputs.each do |input|
-          input.add_observer counter
-        end
+      inputs.each do |input|
+        input.add_observer counter
       end
-
-      result
     end
 
+    result
   end
 
-  def self.dataflow(*inputs, &block)
-    Dataflow::dataflow(*inputs, &block)
-  end
-
+  module_function :dataflow
 end

md/dataflow.md

@@ -55,7 +55,7 @@ def fib(n)
   end
 end
 
-puts fib(14)
+puts fib(14) #=> 377
 ```
 
 We could modify this to use futures.
@@ -71,10 +71,12 @@ def fib(n)
   end
 end
 
-f = fib(14)
-f.execute
+f = fib(14) #=> #<Concurrent::Future:0x000001019ef5a0 ...
+f.execute   #=> #<Concurrent::Future:0x000001019ef5a0 ...
+
 sleep(0.5)
-puts f.value
+
+puts f.value #=> 377
 ```
 
 One of the drawbacks of this approach is that all the futures start, and then
@@ -128,9 +130,10 @@ end
 We can wrap this up in a dataflow utility.
 
 ```ruby
-f = fib(14)
+f = fib(14) #=> #<Concurrent::Future:0x00000101fca308 ...
 sleep(0.5)
-puts f.value
+
+puts f.value #=> 377
 
 def dataflow(*inputs, &block)
   result = Concurrent::Future.new(&block)
@@ -158,9 +161,10 @@ def fib(n)
   end
 end
 
-f = fib(14)
+f = fib(14) #=> #<Concurrent::Future:0x00000101fca308 ...
 sleep(0.5)
-puts f.value
+
+puts f.value #=> 377
 ```
 
 Since we know that the futures the dataflow computation depends on are already
@@ -179,7 +183,7 @@ def dataflow(*inputs, &block)
   if inputs.empty?
     result.execute
   else
-    barrier = Concurrent::CountingObserver.new(inputs.size) { result.execute }
+    barrier = CountingObserver.new(inputs.size) { result.execute }
 
     inputs.each do |input|
       input.add_observer barrier
@@ -189,7 +193,8 @@ def dataflow(*inputs, &block)
   result
 end
 
-f = fib(14)
+f = fib(14) #=> #<Concurrent::Future:0x000001019a26d8 ...
 sleep(0.5)
-puts f.value
+
+puts f.value #=> 377
 ```