OutputStream that writes to a scalaz-stream mutable Queue

QueueOutputStream.scala

import java.io.OutputStream

import scalaz.stream._

import scodec.bits.ByteVector

class QueueOutputStream(q: async.mutable.Queue[ByteVector]) extends OutputStream {
  override def write(b: Int): Unit =
    q.enqueueOne(ByteVector(b & 0xff)).run

  override def write(b: Array[Byte], off: Int, len: Int): Unit = {
    val bv = ByteVector.view(b).drop(off).take(len).copy
    q.enqueueOne(bv).run
  }

  override def close(): Unit =
    q.close.run

  // feels dangerous. Perhaps it should be a no-op instead.
  override def flush(): Unit =
    q.size.discrete.find(_ == 0).run.run
}

Use with q.dequeue, q.dequeueBatch, or q.dequeueAvailable as a source to a scalaz-stream pipeline.