Register and ignore task exceptions in scalaz threads

Question

Take an example from some floating point documents, but with a theoretical twist.

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 

In this case, the file may well contain some non-dual string, and fahrenheitToCelsius will raise some NumberFormatException . Let's say that in this case we want, perhaps, to record this error and ignore it for further processing of the stream. What is the idiomatic way to do this? I saw a few examples, but they usually collectFrom stream.

Answer 1

You can do this with scalaz. \ / and additional processing steps

  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