scala - 在没有错误状态的情况下处理 iteratee 库中的异常

Question

我正在尝试编写一个枚举器，用于使用 Scalaz 从 java.io.BufferedReader 逐行读取文件7 的 iteratee 库，目前只为 java.io.Reader 提供一个(非常慢的)枚举器。

我遇到的问题与我使用过的所有其他 iteratee 库(例如 Haskell 的 Play 2.0's 和 John Millikin's enumerator)都有一个错误状态，因为它们的 Step 类型的构造函数，而 Scalaz 7 没有。

我当前的实现

这是我目前拥有的。首先是一些导入和 IO 包装器:

import java.io.{ BufferedReader, File, FileReader }
import scalaz._, Scalaz._, effect.IO, iteratee.{ Iteratee => I, _ }

def openFile(f: File) = IO(new BufferedReader(new FileReader(f)))
def readLine(r: BufferedReader) = IO(Option(r.readLine))
def closeReader(r: BufferedReader) = IO(r.close())

还有一个类型别名来清理一些东西:

type ErrorOr[A] = Either[Throwable, A]

现在是一个 tryIO 帮助器，以 enumerator 中的一个为模型(松散地，可能是错误的):

def tryIO[A, B](action: IO[B]) = I.iterateeT[A, IO, ErrorOr[B]](
  action.catchLeft.map(
    r => I.sdone(r, r.fold(_ => I.eofInput, _ => I.emptyInput))
  )
)

BufferedReader 本身的枚举器:

def enumBuffered(r: => BufferedReader) = new EnumeratorT[ErrorOr[String], IO] {
  lazy val reader = r
  def apply[A] = (s: StepT[ErrorOr[String], IO, A]) => s.mapCont(k =>
    tryIO(readLine(reader)) flatMap {
      case Right(None)       => s.pointI
      case Right(Some(line)) => k(I.elInput(Right(line))) >>== apply[A]
      case Left(e)           => k(I.elInput(Left(e)))
    }
  )
}

最后是一个负责打开和关闭阅读器的枚举器:

def enumFile(f: File) = new EnumeratorT[ErrorOr[String], IO] {
  def apply[A] = (s: StepT[ErrorOr[String], IO, A]) => s.mapCont(k =>
    tryIO(openFile(f)) flatMap {
      case Right(reader) => I.iterateeT(
        enumBuffered(reader).apply(s).value.ensuring(closeReader(reader))
      )
      case Left(e) => k(I.elInput(Left(e)))
    }
  )
}

现在假设我想将文件中至少包含二十五个 '0' 字符的所有行收集到一个列表中。我可以写:

val action: IO[ErrorOr[List[String]]] = (
  I.consume[ErrorOr[String], IO, List] %=
  I.filter(_.fold(_ => true, _.count(_ == '0') >= 25)) &=
  enumFile(new File("big.txt"))
).run.map(_.sequence)

在很多方面，这似乎工作得很好:我可以使用 unsafePerformIO 开始操作，它将在几分钟内将数千万行和千兆字节的数据分块到恒定内存中并且不会破坏堆栈，然后在完成后关闭阅读器。如果我给它一个不存在的文件名，它会尽职尽责地给我返回包含在 Left 中的异常，并且 enumBuffered 至少看起来表现得合适如果在读取时遇到异常。

潜在问题

不过，我对我的实现有一些担忧，尤其是 tryIO。例如，假设我尝试编写一些迭代器:

val it = for {
  _ <- tryIO[Unit, Unit](IO(println("a")))
  _ <- tryIO[Unit, Unit](IO(throw new Exception("!")))
  r <- tryIO[Unit, Unit](IO(println("b")))
} yield r

如果我运行它，我会得到以下结果:

scala> it.run.unsafePerformIO()
a
b
res11: ErrorOr[Unit] = Right(())

如果我在 GHCi 中使用 enumerator 尝试同样的操作，结果会更像我所期望的:

...> run $ tryIO (putStrLn "a") >> tryIO (error "!") >> tryIO (putStrLn "b")
a
Left !

我只是没有找到一种方法来获得这种行为，而不会在 iteratee 库本身中出现错误状态。

我的问题

我并不声称自己是迭代器方面的专家，但我在一些项目中使用过各种 Haskell 实现，感觉我或多或少理解了基本概念，并且与 Oleg 喝过一次咖啡。不过，我在这里不知所措。这是在没有错误状态的情况下处理异常的合理方法吗？有没有一种方法可以实现 tryIO ，其行为更像 enumerator 版本？由于我的实现行为不同，是否有某种定时炸弹在等着我？

Answer 1

编辑这里是真正的解决方案。我留在原来的帖子中是因为我认为值得一看的模式。适用于 Klesli 的方法适用于 IterateeT

import java.io.{ BufferedReader, File, FileReader }
import scalaz._, Scalaz._, effect._, iteratee.{ Iteratee => I, _ }

object IterateeIOExample {
  type ErrorOr[+A] = EitherT[IO, Throwable, A]

  def openFile(f: File) = IO(new BufferedReader(new FileReader(f)))
  def readLine(r: BufferedReader) = IO(Option(r.readLine))
  def closeReader(r: BufferedReader) = IO(r.close())

  def tryIO[A, B](action: IO[B]) = I.iterateeT[A, ErrorOr, B] {
    EitherT.fromEither(action.catchLeft).map(r => I.sdone(r, I.emptyInput))
  }

  def enumBuffered(r: => BufferedReader) = new EnumeratorT[String, ErrorOr] {
    lazy val reader = r
    def apply[A] = (s: StepT[String, ErrorOr, A]) => s.mapCont(k =>
      tryIO(readLine(reader)) flatMap {
        case None => s.pointI
        case Some(line) => k(I.elInput(line)) >>== apply[A]
      })
  }

  def enumFile(f: File) = new EnumeratorT[String, ErrorOr] {
    def apply[A] = (s: StepT[String, ErrorOr, A]) => 
      tryIO(openFile(f)).flatMap(reader => I.iterateeT[String, ErrorOr, A](
        EitherT(
          enumBuffered(reader).apply(s).value.run.ensuring(closeReader(reader)))))
  }

  def main(args: Array[String]) {
    val action = (
      I.consume[String, ErrorOr, List] %=
      I.filter(a => a.count(_ == '0') >= 25) &=
      enumFile(new File(args(0)))).run.run

    println(action.unsafePerformIO().map(_.size))
  }
}

=====原帖=====

我觉得你需要一个 EitherT 来混合。如果没有 EitherT，你最终只会得到 3 个左或右。有了 EitherT，它就会把左权占为己有。

我认为你真正想要的是

type ErrorOr[+A] = EitherT[IO, Throwable, A] 
I.iterateeT[A, ErrorOr, B]

以下代码模仿您当前编写内容的方式。因为 IterateeT 没有左和右的概念，所以当你组合它时，你最终只会得到一堆 IO/Id。

scala> Kleisli((a:Int) => 4.right[String].point[Id])
res11: scalaz.Kleisli[scalaz.Scalaz.Id,Int,scalaz.\/[String,Int]] = scalaz.KleisliFunctions$$anon$18@73e771ca

scala> Kleisli((a:Int) => "aa".left[Int].point[Id])
res12: scalaz.Kleisli[scalaz.Scalaz.Id,Int,scalaz.\/[String,Int]] = scalaz.KleisliFunctions$$anon$18@be41b41

scala> for { a <- res11; b <- res12 } yield (a,b)
res15: scalaz.Kleisli[scalaz.Scalaz.Id,Int,(scalaz.\/[String,Int], scalaz.\/[String,Int])] = scalaz.KleisliFunctions$$anon$18@42fd1445

scala> res15.run(1)
res16: (scalaz.\/[String,Int], scalaz.\/[String,Int]) = (\/-(4),-\/(aa))

在下面的代码中，我们不使用 Id，而是使用 EitherT。由于 EitherT 与 Either 具有相同的绑定(bind)行为，因此我们最终得到了我们想要的结果。

scala>  type ErrorOr[+A] = EitherT[Id, String, A]
defined type alias ErrorOr

scala> Kleisli[ErrorOr, Int, Int]((a:Int) => EitherT(4.right[String].point[Id]))
res22: scalaz.Kleisli[ErrorOr,Int,Int] = scalaz.KleisliFunctions$$anon$18@58b547a0

scala> Kleisli[ErrorOr, Int, Int]((a:Int) => EitherT("aa".left[Int].point[Id]))
res24: scalaz.Kleisli[ErrorOr,Int,Int] = scalaz.KleisliFunctions$$anon$18@342f2ceb

scala> for { a <- res22; b <- res24 } yield 2
res25: scalaz.Kleisli[ErrorOr,Int,Int] = scalaz.KleisliFunctions$$anon$18@204eab31

scala> res25.run(2).run
res26: scalaz.Scalaz.Id[scalaz.\/[String,Int]] = -\/(aa)

您可以将 Keisli 替换为 IterateeT，将 Id 替换为 IO 以获得您需要的内容。