streamer2.js

function stream(next, end) {
  next(1)
  next(2)
  next(3)
  end()
}

[1, 2, 3, 4] -> take(3) -> map(inc) -> take(2)

function map(f, input) {
  return function mapped(next, end) {
    input(function(value) {
      return next(f(value))
    }, end)
  }
}

function filter(p, input) {
  return function filtered(next, end) {
    input(function(value) {
      return p(value) ? next(value) : value
    }, end)
  }
}

function reductions(f, initial, input) {
  var nil = {}
  var source = input || initial
  var start = input ? initial : nil
  return function reduced(next, end) {
    var result = start
    source(function(value) {
      result = result === nil ? value : f(result, value)
      return next(result)
    }, end)
  }
}

function last(input) {
  var nil = {}
  return function last(next, end) {
    var result = nil
    source(function(value) {
      result = value
    }, function(error) {
      if (result !== nil) next(result)
      end(error)
    })
  }
}

function reduce(f, initial, input) {
  return last(reductions(f, initial, input))
}

function stop() { throw Error("Stop!") }
function empty(_, end) { end() }

function take(n, input) {
  return n <= 0 ? empty : function taken(next, end) {
    var left = n
    input(function(value) {
      left = left - 1
      var result = next(value)
      return left === 0 ? stop : result
    }, end)
  }
}

function takeWhile(p, input) {
  return function taken(next, end) {
    input(function(value) {
      return p(value) ? next(value) : stop
    }, end)
  }
}

function drop(n, input) {
  return n <= 0 ? input : function dropped(next, end) {
    var left = n + 1
    input(function(value) {
      left = left ? left - 1 : 0
      return left ? value : next(value)
    }, end)
  }
}

function dropWhile(p, input) {
  return function dropped(next, end) {
    input(function(value) {
      return p(value) ? value : next(value)
    }, end)
  }
}

function merge(inputs) {
  return function merged(next, end) {
    var count = 1
    var error = null
    var state = null
    
    function close(reason) {
      count = count - 1
      if (!error && opened === 0) end(reason)
      error = error || reason
    }
    
    function open(input) {
      count = count + 1
      // TODO: If backpressure has being applied should merge
      // apply to all the inputs, or just one that send a value ?
      // If it should be applied to all how do we make it work
      // without specialcasing backpressure ?
      return state === stop ? stop : input(forward, close)
    }
    
    function forward(value) {
      state = state === stop ? stop : next(value)
      return state
    }
    
    inputs(function(input) {
      return error ? stop : open(input)
    }, close)
  }
}