var Stream = require('stream')
  
  // through
  //
  // a stream that does nothing but re-emit the input.
  // useful for aggregating a series of changing but not ending streams into one stream)
  
  exports = module.exports = through
  through.through = through
  
  //create a readable writable stream.
  
  function through (write, end, opts) {
    write = write || function (data) { this.queue(data) }
    end = end || function () { this.queue(null) }
  
    var ended = false, destroyed = false, buffer = [], _ended = false
    var stream = new Stream()
    stream.readable = stream.writable = true
    stream.paused = false
  
  //  stream.autoPause   = !(opts && opts.autoPause   === false)
    stream.autoDestroy = !(opts && opts.autoDestroy === false)
  
    stream.write = function (data) {
      write.call(this, data)
      return !stream.paused
    }
  
    function drain() {
      while(buffer.length && !stream.paused) {
        var data = buffer.shift()
        if(null === data)
          return stream.emit('end')
        else
          stream.emit('data', data)
      }
    }
  
    stream.queue = stream.push = function (data) {
  //    console.error(ended)
      if(_ended) return stream
      if(data === null) _ended = true
      buffer.push(data)
      drain()
      return stream
    }
  
    //this will be registered as the first 'end' listener
    //must call destroy next tick, to make sure we're after any
    //stream piped from here.
    //this is only a problem if end is not emitted synchronously.
    //a nicer way to do this is to make sure this is the last listener for 'end'
  
    stream.on('end', function () {
      stream.readable = false
      if(!stream.writable && stream.autoDestroy)
        process.nextTick(function () {
          stream.destroy()
        })
    })
  
    function _end () {
      stream.writable = false
      end.call(stream)
      if(!stream.readable && stream.autoDestroy)
        stream.destroy()
    }
  
    stream.end = function (data) {
      if(ended) return
      ended = true
      if(arguments.length) stream.write(data)
      _end() // will emit or queue
      return stream
    }
  
    stream.destroy = function () {
      if(destroyed) return
      destroyed = true
      ended = true
      buffer.length = 0
      stream.writable = stream.readable = false
      stream.emit('close')
      return stream
    }
  
    stream.pause = function () {
      if(stream.paused) return
      stream.paused = true
      return stream
    }
  
    stream.resume = function () {
      if(stream.paused) {
        stream.paused = false
        stream.emit('resume')
      }
      drain()
      //may have become paused again,
      //as drain emits 'data'.
      if(!stream.paused)
        stream.emit('drain')
      return stream
    }
    return stream
  }