oneToOne.js

var oneToOne = function(innerWrite, innerEnd) {
  var stream = through(write, end)
    , pending = 0
    , queue
    , ended
    , abort

  queue = stream.queue
  stream.queue = _queue

  return stream

  function _queue(what) {
    --pending

    if(what === null) {
      abort = true
    }

    queue.call(stream, what)

    return check()
  }

  function write(datum) {
    ++pending
    innerWrite.call(stream, datum)
  }

  function end() {
    ended = true
    check()
  }

  function check() {
    if(pending || abort) {
      return
    }

    innerEnd.call(stream)

    if(abort) {
      return
    }

    queue.call(stream, null)
  }
}