function collect(cb) {
  var ary = []
  return {
    write: ary.push.bind(ary),
    paused: false,
    end: function (err) {
      this.ended = err || true
      cb(err, ary)
    }
  }
}

function values (ary) {
  var i = 0
  return {
    resume: function () {
      //just loop until the sink pauses or we have ended
      while(!this.sink.paused && !this.ended)
        if(this.ended = i >= ary.length) this.sink.end()
        else                             this.sink.write(ary[i++])
    },
    sink: null,
    pipe: pipe
  }
}

function pipe(sink) {
  this.sink = sink
  sink.source = this
  if(!sink.paused) this.resume()
  return sink
}

function map(fn) {
  return {
    write: function (data) {
      //we trust the source not to call us if we are paused
      //so we don't need to buffer this.
      this.sink.write(fn(data))
      this.paused = this.sink.paused //update our pause state if it changed
    },
    end: function (err) {
      this.ended = true
      this.sink.end(err)
    },
    //a map needs to start off paused
    paused: true,
    //the sink will call resume when it unpauses.
    resume: function () {
      if(!(this.paused = this.sink.paused))
        this.source.resume()
    },
    pipe: pipe,
    source: null, sink: null
  }
}

function asyncMap (asyncFn) {
  var _map = map()
  _map.write = function (data) {
    //we trust the source not to call us if we are paused
    //so we don't need to buffer this.
    var self = this
    self.paused = true
    asyncFn(data, function (err, mapped_data) {
      self.paused = false
      if(err) self.sink.end(this.ended = err)
      else {
        self.sink.write(mapped_data)
        self.resume()
      }
    })
  }
  return _map
}

function values (ary) {
  var i = 0
  return {
    ...
    abort: function (err) {
      this.ended = true
      this.sink.end(err)
    }
  }
}

function map(fn) {
  return {
    ...
    abort: function (err) {
      this.source.abort(err)
    }
  }
}

> node pull-stream/benchmarks/node.js
pull3*100000: 4470.156ms
> node pull-stream/benchmarks/pull.js
pull3*100000: 640.037ms
> node push-stream/benchmarks/push.js
pull3*100000: 226.907ms