Computing the MD5 hash of a file and writing it using scalaz-stream

md5-stream.scala

import scalaz.stream.{Process1, Process, io, process1}
import Process._
import java.security._
import scalaz._
import scala.collection.mutable
import java.math.BigInteger


  val P = Process

  // store the final digest in a buffer
  val digestBuffer = new ValueBuffer[MessageDigest]
  val p1 =
    Process.range(0, 10)
      .map(_.toString)
      .pipe(process1.utf8Encode)
      // use a writer to compute the current MD5 digest
      .pipe(logMd5)
      // "drain" the Write values to a buffer just keeping the last one
      .drainW(io.fillBuffer(digestBuffer))
      .to(io.fileChunkW("target/test"))

  // run the process to write the file
  p1.runLast.run
  // print the MD5
  println(new BigInteger(1, digestBuffer.toList.head.digest).toString(16))

  // Process1 updating a MD5 digest when receiving Array[Byte]
  def logMd5: Process1[Array[Byte], MessageDigest \/ Array[Byte]] = {

    def go(digest: MessageDigest): Process1[Array[Byte], MessageDigest \/ Array[Byte]] = {
      receive1 {
        case arr =>
          digest.update(arr)
          (P.emitO(arr) ++ P.emitW(digest)) fby go(digest)
      }
    }
    go(MessageDigest.getInstance("MD5"))
  }

  // One value max buffer
  class ValueBuffer[A] extends mutable.Buffer[A] {
    private var element: Option[A] = None

    def +=(elem: A) = { element = Some(elem); this }
    def +=:(elem: A)= { element = Some(elem); this }
    def apply(n: Int): A = ???
    def clear(): Unit = element = None
    def insertAll(n: Int,elems: Traversable[A]): Unit = ???
    def length: Int = element.toList.size
    def remove(n: Int): A = ???
    def update(n: Int,newelem: A): Unit = ???
    def iterator: Iterator[A] = element.toList.iterator
  }

}

Here is what I had in mind:

val p1 = Process.range(0, 10)
  .map(_.toString)
  .pipe(process1.utf8Encode)

val digest = Process.logged(p1)
  .pipeO(md5)
  .drainW(io.fileChunkW("target/test"))
  .runLast.run

Oh my... this is embarrassing. The problem with pipeO can be obviated if we drainW to the sink and then pipe to md5:

val digest = Process.logged(p1)
  .drainW(io.fileChunkW("target/test"))
  .pipe(md5)
  .runLast.run

Eric, does this work for you?