Merge branch 'preread_reduce'

sinks/reduce.js

@@ -2,31 +2,21 @@
 
 var drain = require('./drain')
 
-module.exports = function reduce (/* reducer, acc, cb */) {
-  var reducer = arguments[0]
-  var acc
-  var cb
-  if (arguments.length === 3) {
-    acc = arguments[1]
-    cb = arguments[2]
-    return drain(function (data) {
-      acc = reducer(acc, data)
-    }, function (err) {
-      cb(err, acc)
-    })
-  } else {
-    cb = arguments[1]
-    var first = true
-    return drain(function (data) {
-      if (first) {
-        acc = data
-        first = false
-      } else {
-        acc = reducer(acc, data)
-      }
-    }, function (err) {
-      cb(err, acc)
-    })
-  }
+module.exports = function reduce (reducer, acc, cb ) {
+  if(!cb) cb = acc, acc = null
+  var sink = drain(function (data) {
+    acc = reducer(acc, data)
+  }, function (err) {
+    cb(err, acc)
+  })
+  if (arguments.length === 2)
+    return function (source) {
+      source(null, function (end, data) {
+        //if ended immediately, and no initial...
+        if(end) return cb(end === true ? null : end)
+        acc = data; sink(source)
+      })
+    }
+  else
+    return sink
 }
-