From af9fe76e3ffe2b1f35024b5060fdec9bd3377f49 Mon Sep 17 00:00:00 2001
From: Dominic Tarr <dominic.tarr@gmail.com>
Date: Wed, 7 Sep 2011 15:28:18 +1000
Subject: [PATCH] add parallel and serial map functions

---
 ctrlflow.group.asynct.js |  19 ++++++++
 ctrlflow.map.asynct.js   | 102 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 121 insertions(+)
 create mode 100644 ctrlflow.map.asynct.js

diff --git a/ctrlflow.group.asynct.js b/ctrlflow.group.asynct.js
index d05c470..52d71bc 100644
--- a/ctrlflow.group.asynct.js
+++ b/ctrlflow.group.asynct.js
@@ -153,3 +153,22 @@ exports['group can be seq array - edgecase'] = function (test){
     test.done()
   })  
 }
+
+exports ['each grouped step gets the same args'] = function (test) {
+  var args = [1,true, 25.34, {}, [0]]
+  ctrl.group({
+    A: d.fallthrough, //fallthrough passes args straight to callback.
+    B: d.fallthrough,
+    C: d.fallthrough,
+    })
+  (1,true, 25.34, {}, [0], function (err, results) {
+    
+    it(results).has ({
+      A: [1,true, 25.34, {}, [0]],
+      B: [1,true, 25.34, {}, [0]],
+      C: [1,true, 25.34, {}, [0]],
+    })
+    test.done()  
+  })
+
+}
diff --git a/ctrlflow.map.asynct.js b/ctrlflow.map.asynct.js
new file mode 100644
index 0000000..e6ed099
--- /dev/null
+++ b/ctrlflow.map.asynct.js
@@ -0,0 +1,102 @@
+
+var ctrl = require('ctrlflow')
+  , it = require('it-is')
+  , d = require('d-utils')
+  , fs = require('fs')
+  , path = require('path')
+
+exports['simple serial'] = simple(ctrl.serial)
+exports['simple parallel'] = simple(ctrl.parallel)
+
+function simple(type) {
+  return function (test) {
+    //this map will map to the input through next tick
+    var called = 0
+    var mapper = 
+      type.map(function (value, key, callback) {
+        called ++
+  //      /*d.delay*/(d.fallthrough)(value, callback) //  
+        callback(null, value)
+      }, true)
+
+    it(mapper).function()
+
+    mapper('abcdef'.split(''), function (err, results) {
+      if(err) throw err
+      it(results.join('')).equal('abcdef')
+      test.done()
+    })
+
+  }
+}
+exports['ls parallel']  = ls(ctrl.parallel)
+exports['ls serial']    = ls(ctrl.serial)
+function ls (type) {
+  return function (test) {
+    var ls
+
+    ctrl([
+      fs.readdir
+    , function (data, callback) {
+        ls = data
+        callback(null,data)
+      }
+    , type.map(fs.stat)
+    ])
+    (__dirname, function (err, data) {
+      if(err) throw err
+      it(data.length).equal(ls.length)
+      test.done()
+   })
+  }
+}
+
+exports['ls-r parallel']  = lsR(ctrl.parallel)
+exports['ls-r serial']    = lsR(ctrl.serial)
+function lsR(type) {
+  return function (test) {
+    //fs has an arkward interface, because file are objects (state + behaviour) but fs is functions
+
+    var ls = 
+    ctrl([
+      function (fn, cb) {
+        console.log('dir:', fn)
+        fs.stat(fn, function (err, stat) {
+          if(err)
+            cb(err)
+          else {
+            stat.name = fn
+            cb(null, stat)
+          }
+        })
+      }
+    , function (file, next) {
+        if(file.isDirectory())
+          ctrl([
+            [fs.readdir, file.name]
+          , function (fn,next) {
+              next(null, d.map(fn, function (v) {return path.join(file.name, v)}))
+            }
+          , type.map(ls)
+          ])(next)
+        else
+          next(null, file)
+      }
+    ])
+
+    ls(__dirname, function (err, data) {
+      if(err)
+        throw err 
+      function print(file) {
+        if(Array.isArray(file))
+          d.each(file, function (f) {
+            print(f)
+          })
+        else
+          console.log(file.name)
+      }
+      print(data)
+      test.done()
+    })
+  }
+}
\ No newline at end of file