试图理解 Scala 枚举器/迭代器

Question

我是 Scala 和 Play! 的新手，但在使用 Django 和 Python 构建 web 应用程序以及一般编程方面有相当多的经验。

我一直在做自己的练习以试图提高我的理解力 - 只需从数据库中提取一些记录并将它们输出为 JSON 数组即可。我正在尝试使用 Enumarator/Iteratee 功能来执行此操作。

我的代码如下：

TestObjectController.scala：

  def index = Action {
    db.withConnection { conn=>
      val stmt = conn.createStatement()
      val result = stmt.executeQuery("select * from datatable")

      logger.debug(result.toString)
      val resultEnum:Enumerator[TestDataObject] = Enumerator.generateM {
        logger.debug("called enumerator")
        result.next() match {
          case true =>
            val obj = TestDataObject(result.getString("name"), result.getString("object_type"),
              result.getString("quantity").toInt, result.getString("cost").toFloat)
            logger.info(obj.toJsonString)
            Future(Some(obj))
          case false =>
            logger.warn("reached end of iteration")
            stmt.close()
            null
        }
      }

      val consume:Iteratee[TestDataObject,Seq[TestDataObject]] = {
        Iteratee.fold[TestDataObject,Seq[TestDataObject]](Seq.empty[TestDataObject]) { (result,chunk) => result :+ chunk }
      }

      val newIteree = Iteratee.flatten(resultEnum(consume))
      val eventuallyResult:Future[Seq[TestDataObject]] = newIteree.run
      eventuallyResult.onSuccess { case x=> println(x)}
      Ok("")

    }
  }

TestDataObject.scala：

package models

case class TestDataObject (name: String, objtype: String, quantity: Int, cost: Float){
  def toJsonString: String = {
    val mapper = new ObjectMapper()
    mapper.registerModule(DefaultScalaModule)
    mapper.writeValueAsString(this)
  }
}

我有两个主要问题：

• 如何通过 Enumerator 回调指示输入已完成？文档说“这个方法需要一个回调函数 e: => Future[Option[E]] ，每次应用这个 Enumerator 的迭代器准备好接受一些输入时都会调用它。”但我无法传递我发现的任何类型的 EOF，因为它是错误的类型。将其包装在 Future 中没有帮助，但本能地我不确定这是正确的方法。

• 如何从 Future 中获取最终结果以从控制器视图中返回？我的理解是，我实际上需要暂停主线程以等待子线程完成，但我见过的唯一示例以及我在未来课程中发现的唯一内容是 onSuccess 回调 - 但我该怎么做从视图中返回？ Iteratee.run 是否会阻塞直到所有输入都被消耗完？

还有几个子问题，以帮助我理解：

• 当我的对象已经在 Future 中时，为什么我需要将它包装在 Some() 中？ Some() 究竟代表什么？
• 当我第一次运行代码时，我从 logger.info 中记录了一条记录，然后它报告“到达迭代结束”。随后在同一个会话中运行什么都不调用。不过，我正在关闭声明，那么为什么我第二次没有得到任何结果呢？我原以为它会无限循环，因为我不知道如何发出正确的循环终止信号。

非常感谢任何答案，我以为我已经掌握了这个窍门，但显然还没有！

Answer 1

如何通过 Enumerator 回调指示输入已完成？

你返回一个Future(None)。

如何让 Future 的最终结果从控制器视图返回？

您可以使用Action.async (doc)：

def index = Action.async {
  db.withConnection { conn=>
    ...
    val eventuallyResult:Future[Seq[TestDataObject]] = newIteree.run
    eventuallyResult map { data =>
      OK(...)
    }
  }
}

当我的对象已经在 Future 中时，为什么我需要将它包装在 Some() 中？ Some() 究竟代表什么？

Future 表示获取下一个元素的（可能是异步的）处理。 Option 表示下一个元素的可用性：Some(x) 如果另一个元素可用，None 如果枚举完成。