在Scala中的异常上重新启动迭代器

Question

我有一个迭代器（实际上是 Source.getLines），它从 URL 读取无限的数据流。当出现连接问题时，迭代器偶尔会抛出java.io.IOException。在这种情况下，我需要重新连接并重新启动迭代器。我希望这是无缝的，这样迭代器对消费者来说就像一个普通的迭代器，但在必要时会在下面重新启动。

例如，我希望看到以下行为：

scala> val iter = restartingIterator(() => new Iterator[Int]{
  var i = -1
  def hasNext = {
    if (this.i < 3) {
      true
    } else {
      throw new IOException
    }
  }
  def next = {
    this.i += 1
    i
  }
})
res0: ...

scala> iter.take(6).toList
res1: List[Int] = List(0, 1, 2, 3, 0, 1)

我对这个问题有部分解决方案，但它会在某些极端情况下失败（例如，重新启动后第一个项目上的 IOException）并且非常难看：

def restartingIterator[T](getIter: () => Iterator[T]) = new Iterator[T] {
  var iter = getIter()
  def hasNext = {
    try {
      iter.hasNext
    } catch {
      case e: IOException => {
        this.iter = getIter()
        iter.hasNext
      }
    }
  }
  def next = {
    try {
      iter.next
    } catch {
      case e: IOException => {
        this.iter = getIter()
        iter.next
      }
    }
  }
}

我一直觉得有更好的解决方案，可能是 Iterator.continually 和 util.control.Exception 的某种组合或类似的东西，但我想不出一个。有什么想法吗？

Answer 1

这与您的版本相当接近，并使用scala.util.control.Exception：

def restartingIterator[T](getIter: () => Iterator[T]) = new Iterator[T] {
  import util.control.Exception.allCatch
  private[this] var i = getIter()
  private[this] def replace() = i = getIter()
  def hasNext: Boolean = allCatch.opt(i.hasNext).getOrElse{replace(); hasNext}
  def next(): T = allCatch.opt(i.next).getOrElse{replace(); next}
}

由于某种原因，这不是尾递归，但可以通过使用稍微详细一点的版本来解决：

def restartingIterator2[T](getIter: () => Iterator[T]) = new Iterator[T] {
  import util.control.Exception.allCatch
  private[this] var i = getIter()
  private[this] def replace() = i = getIter()
  @annotation.tailrec def hasNext: Boolean = {
    val v = allCatch.opt(i.hasNext)
    if (v.isDefined) v.get else {replace(); hasNext}
  }
  @annotation.tailrec def next(): T = {
    val v = allCatch.opt(i.next)
    if (v.isDefined) v.get else {replace(); next}
  }
}

---

编辑：有util.control.Exception和Iterator.continually的解决方案：

def restartingIterator[T](getIter: () => Iterator[T]) = {
  import util.control.Exception.allCatch
  var iter = getIter()
  def f: T = allCatch.opt(iter.next).getOrElse{iter = getIter(); f}
  Iterator.continually { f }
}

Answer 2

有一个更好的解决方案，Iteratee：

http://apocalisp.wordpress.com/2010/10/17/scalaz-tutorial-enumeration-based-io-with-iteratees/

这里是一个枚举器，遇到异常会重新启动。

def enumReader[A](r: => BufferedReader, it: IterV[String, A]): IO[IterV[String, A]] = {
  val tmpReader = r
  def loop: IterV[String, A] => IO[IterV[String, A]] = {
    case i@Done(_, _) => IO { i }
    case Cont(k) => for {
      s <- IO { try { val x = tmpReader.readLine; IO(x) }
                catch { case e => enumReader(r, it) }}.join
      a <- if (s == null) k(EOF) else loop(k(El(s)))
    } yield a
  }
  loop(it)
}

内部循环推进 Iteratee，但外部函数仍保留原始函数。由于Iteratee是一个持久化的数据结构，要重新启动你只需要再次调用该函数。

我在这里按名称传递 Reader，以便 r 本质上是一个为您提供全新（重新启动）阅读器的函数。在实践中，您将希望更有效地将其括起来（关闭现有的异常读者）。

Answer 3

这是一个不起作用的答案，但感觉应该：

def restartingIterator[T](getIter: () => Iterator[T]): Iterator[T] = {
  new Traversable[T] {
    def foreach[U](f: T => U): Unit = {
      try {
        for (item <- getIter()) {
          f(item)
        }
      } catch {
        case e: IOException => this.foreach(f)
      }
    }
  }.toIterator
}

我认为这非常清楚地描述了控制流，这很棒。

由于bug in Traversable.toStream，此代码将在Scala 2.8.0 中抛出StackOverflowError，但即使修复了该错误，此代码仍然不适用于我的用例，因为toIterator 调用@987654325 @，表示它会将所有项目存储在内存中。

我希望能够通过编写 foreach 方法来定义 Iterator，但似乎没有任何简单的方法可以做到这一点。