在scalaz-streams中记录并忽略Task中的异常

Question

让我们从一些scalaz-stream文档中举例说明,但是有一个理论上的转折.

import scalaz.stream._
import scalaz.concurrent.Task

val converter: Task[Unit] =
  io.linesR("testdata/fahrenheit.txt")
    .filter(s => !s.trim.isEmpty && !s.startsWith("//"))
    .map(line => fahrenheitToCelsius(line.toDouble).toString)
    .intersperse("\n")
    .pipe(text.utf8Encode)
    .to(io.fileChunkW("testdata/celsius.txt"))
    .run

// at the end of the universe...
val u: Unit = converter.run

在这种情况下,文件可能包含一些非双字符串,并且fahrenheitToCelsius会抛出一些NumberFormatException.让我们说在这种情况下,我们希望记录此错误并忽略它以进行进一步的流处理.这样做的惯用方法是什么？我见过一些例子,但他们通常collectFrom是流.

Answer 1

您可以使用 scalaz.\/ 和其他处理步骤来完成此操作

  def fahrenheitToCelsius(line: String): Throwable \/ String = 
     \/.fromTryCatchNonFatal {
        val fahrenheit = line.toDouble
        val celsius = fahrenheit // replace with real convert
        celsius.toString
     }

  def collectSome[T]: PartialFunction[Option[T], T] = {
    case Some(v) => v
  }

  def logThrowable[T]: PartialFunction[Throwable \/ T, Option[T]] = {
    case -\/(err) => 
      err.printStackTrace()
      None
    case \/-(v) => Some(v)
  }

  val converter: Task[Unit] =
    io.linesR("testdata/fahrenheit.txt")
      .filter(s => !s.trim.isEmpty && !s.startsWith("//"))
      .map(fahrenheitToCelsius)
      .map(logThrowable)
      .collect(collectSome)
      .intersperse("\n")
      .pipe(text.utf8Encode)
      .to(io.fileChunkW("testdata/celsius.txt"))
      .run