1 |
|
2 |
|
3 | var drain = require('./drain')
|
4 |
|
5 | module.exports = function reduce (reducer, acc, cb ) {
|
6 | if(!cb) cb = acc, acc = null
|
7 | var sink = drain(function (data) {
|
8 | acc = reducer(acc, data)
|
9 | }, function (err) {
|
10 | cb(err, acc)
|
11 | })
|
12 | if (arguments.length === 2)
|
13 | return function (source) {
|
14 | source(null, function (end, data) {
|
15 | //if ended immediately, and no initial...
|
16 | if(end) return cb(end === true ? null : end)
|
17 | acc = data; sink(source)
|
18 | })
|
19 | }
|
20 | else
|
21 | return sink
|
22 | }
|