decoder.scala

import scalaz.stream.Process
import java.nio.{ByteBuffer,CharBuffer}
import java.nio.charset.Charset

object Decoder {

  type Bytes = Array[Byte]

  def decode(charset: Charset, leftovers: Bytes = Array()): Process.Process1[Bytes, Char] =
    Process.receive1({ data =>
      val in = ByteBuffer.allocate(data.length + leftovers.length)
      in.put(leftovers).put(data)
      in.flip
      val out = CharBuffer.allocate(data.length + 1)
      charset.newDecoder.decode(in, out, false)
      out.flip
      val chars = new Array[Char](out.length)
      out.get(chars)
      if (in.limit > in.position) {
        val bytes = new Bytes(in.limit - in.position)
        in.get(bytes)
        Process.emitSeq(chars, decode(charset, bytes))
      } else {
        Process.emitAll(chars)
      }
    }, Process.fail(new RuntimeException("EOF encountered mid character")))

  def apply(charset: Charset = Charset.forName("UTF-8")) = decode(charset).repeat

}

Example:

constant(2)
  .through(io.fileChunkR("text"))
  .pipe(Decoder())
  .map(_.mkString.getBytes)
  .to(io.fileChunkW("copy"))
  .run