'use strict'
const pull = require('pull-stream')
// wrap pull streams around packet-stream's weird streams.

function once (fn) {
  let done = false
  return (err, val) => {
    if (done) return
    done = true
    fn(err, val)
  }
}

function duplex (weird, _done) {
  const buffer = []
  let ended = false
  let waiting
  let abort

  const done = once((err, v) => {
    if (_done) _done(err, v)
    // deallocate
    weird = null
    _done = null
    waiting = null
    if (abort) abort(err || true, () => {})
  })

  weird.read = function (data, end) {
    ended = ended || end

    if (waiting) {
      const cb = waiting
      waiting = null
      cb(ended, data)
    } else if (!ended) {
      buffer.push(data)
    }

    if (ended) {
      done(ended !== true ? ended : null)
    }
  }

  return {
    source (abort, cb) {
      if (abort) {
        if (weird) weird.write(null, abort)
        cb(abort)
        done(abort !== true ? abort : null)
      } else if (buffer.length) {
        cb(null, buffer.shift())
      } else if (ended) {
        cb(ended)
      } else {
        waiting = cb
      }
    },
    sink (read) {
      if (ended) {
        abort = null
        return read(ended, () => {})
      }
      abort = read
      pull.drain((data) => {
        // TODO: make this should only happen on a UNIPLEX stream.
        if (ended) return false
        weird.write(data)
      }, (err) => {
        if (weird && !weird.writeEnd) {
          weird.write(null, err || true)
        }
        if (done) done(err)
      })(read)
    }
  }
}

function uniplex (s, done) {
  return duplex(s, (err) => {
    if (!s.writeEnd) s.write(null, err || true)
    if (done) done(err)
  })
}

function source (s) {
  return uniplex(s).source
}

function sink (s, done) {
  return uniplex(s, done).sink
}

module.exports = duplex
module.exports.source = source
module.exports.sink = sink
module.exports.duplex = duplex

@@ -301,16 +301,38 @@
 = false%0A
+  let aborted = false%0A
   let wa
@@ -702,24 +702,36 @@
 e if (!ended
+ && !aborted
 ) %7B%0A      bu
@@ -936,24 +936,73 @@
   cb(abort)%0A
+        buffer.length = 0%0A        aborted = true%0A
         done

Drop messages on abort