
var Stream = require('stream')
var pull   = require('pull-stream')

module.exports = duplex

function duplex (reader, read) {
  var cbs = [], input = [], ended
  var s = new Stream()
  s.writable = s.readable = true
  s.write = function (data) {
    if(cbs.length)
      cbs.shift()(null, data)
    else
      input.push(data)
    return !! cbs.length
  }

  s.end = function () {
    ended = true
    if(cbs.length)
      cbs.shift()(true)      
  }

  s.source = pull.addPipe(function (end, cb) {
    if(input.length) {
      cb(null, input.shift())
      if(!input.length)
        s.emit('drain')
    }
    else if(ended)
      cb(ended)
    else
      cbs.push(cb)
  })

  if(reader) reader(s.source)

  var output = [], _cbs = []
  var _ended = false, waiting = false

  s.sink = function (_read) {
    read = _read
    process.nextTick(drain)
  }

  if(read) s.sink(read)

  function drain () {
    waiting = false
    if(!read) return
    while(output.length && !s.paused) {
      s.emit('data', output.shift())
    }
    if(s.paused) return
    if(_ended)
      return s.emit('end')

    read(null, function next (end, data) {
      console.log('R', end, data)
      if(s.paused) {
        if(end) _ended = end
        else output.push(data)
        waiting = true
      } else {
        if(end) s.emit('end')
        else {
          console.log('data', data)
          s.emit('data', data)
          read(null, next)
        }
      }
    })
  }

  s.pause = function () {
    s.paused = true
    return s
  }

  s.resume = function () {
    s.paused = false
    if(waiting) drain()
    return s
  }

  return s
}


var Stream = require('stream')
var pull   = require('pull-stream')

module.exports = duplex

function duplex (reader, read) {
  var cbs = [], input = [], ended
  var s = new Stream()
  s.writable = s.readable = true
  s.write = function (data) {
    if(cbs.length)
      cbs.shift()(null, data)
    else
      input.push(data)
    return !! cbs.length
  }

  s.end = function () {
    ended = true
    if(cbs.length)
      cbs.shift()(true)      
  }

  s.source = pull.addPipe(function (end, cb) {
    if(input.length) {
      cb(null, input.shift())
      if(!input.length)
        s.emit('drain')
    }
    else if(ended)
      cb(ended)
    else
      cbs.push(cb)
  })

  if(reader) reader(s.source)

  var output = [], _cbs = []
  var _ended = false, waiting = false

  s.sink = function (_read) {
    read = _read
    process.nextTick(drain)
  }

  if(read) s.sink(read)

  function drain () {
    waiting = false
    if(!read) return
    while(output.length && !s.paused) {
      s.emit('data', output.shift())
    }
    if(s.paused) return
    if(_ended)
      return s.emit('end')

    read(null, function next (end, data) {
      if(s.paused) {
        if(end) _ended = end
        else output.push(data)
        waiting = true
      } else {
        if(end) s.emit('end')
        else {
          s.emit('data', data)
          read(null, next)
        }
      }
    })
  }

  s.pause = function () {
    s.paused = true
    return s
  }

  s.resume = function () {
    s.paused = false
    if(waiting) drain()
    return s
  }

  return s
}

remove logging