var pull = require('pull-stream')

var next = 
  'undefined' !== typeof setImmediate 
    ? setImmediate
    : process.nextTick

module.exports = pull.pipeable(function (read, writer, ender) {
  var queue = [], ended
  
  function enqueue (data) {
    queue.push(data)
  }

  writer = writer || function (data) {
    this.queue(data)
  }

  ender = ender || function () {
    this.queue(null)
  }

  var emitter = {
    emit: function (event, data) {
      if(event == 'data') enqueue(data)
      if(event == 'end')  enqueue(null)
    },
    queue: enqueue
  }

  return function (end, cb) {
    ended = ended || end
    ;(function pull () {
      if(queue.length) {
        var data = queue.shift()
        cb(data === null, data)
      } else {
        read(ended, function (end, data) {
           //null has no special meaning for pull-stream
          if(data) writer.call(emitter, data || undefined)
          if(ended = end)  ender.call(emitter)
          next(pull)
        })
      }
    })()
  }
})


@@ -440,24 +440,57 @@
 nt, data) %7B%0A
+      console.error(event, data)%0A
       if(eve
@@ -553,24 +553,64 @@
 queue(null)%0A
+      if(event == 'error') ended = data%0A
     %7D,%0A    q
@@ -714,16 +714,47 @@
 %7B%0A      
+if(ended) cb(ended)%0A      else 
 if(queue
@@ -836,16 +836,22 @@
 %0A      %7D
+%0A     
  else %7B%0A
@@ -1025,24 +1025,33 @@
   if(ended =
+ ended %7C%7C
  end)  ender

handle errors