在没有错误状态的iteratee库中处理异常

Question

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

我遇到的问题与我使用的所有其他iteratee库(例如Play 2.0和John Millikinenumerator for Haskell)都有错误状态作为其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)))
    }
  )
}

现在假设我想将包含至少25个'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得很漂亮:我可以启动它,它将在几分钟内在数千万行和数十亿字节的数据中,在恒定的内存中并且不会烧掉堆栈,然后关闭阅读器当它完成.如果我给它一个不存在的文件的名称,它将尽职尽责地给我回复包含在a中的异常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(())

如果我enumerator在GHCi中尝试相同的事情,结果更像我期望的结果:

...> 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))

在以下代码中,我们使用EitherT而不是使用Id.由于EitherT具有与Either相同的绑定行为,因此我们最终会得到我们想要的内容.

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)

你可以用IterateeT替换Keisli,用IO代替Id来获得你需要的东西.