ProgressInputStream.scala

import java.io.{FilterInputStream, InputStream}

class ProgressInputStream(in: InputStream, listener: Long => Unit) extends FilterInputStream(in) {
  val NotificationThreshold = 8 * 1024;

  var unnotifiedByteCount = 0

  override def read() = {
    val data = super.read()
    if (data != -1) notify(1)
    data
  }

  override def read(b: Array[Byte], off: Int, len: Int) = {
    val bytesRead = super.read(b, off, len)
    if (bytesRead != -1) notify(bytesRead)
    bytesRead
  }

  override def close() {
    if (unnotifiedByteCount > 0) {
      listener(unnotifiedByteCount)
      unnotifiedByteCount = 0
    }

    super.close()
  }

  def notify(bytesRead: Int) {
    unnotifiedByteCount += bytesRead

    if (unnotifiedByteCount >= NotificationThreshold) {
      listener(unnotifiedByteCount)
      unnotifiedByteCount = 0
    }
  }
}

object ProgressInputStream {
  def apply(in: InputStream, availableBytes: Long, tracker: Progress => Unit) =
    new ProgressInputStream(in, new ProgressListener(availableBytes, tracker))
}