如何从 Enumerator/Iteratee 获得结果？答案

【问题标题】：How to get a result from Enumerator/Iteratee?如何从 Enumerator/Iteratee 获得结果？
【发布时间】：2014-08-30 20:01:28
【问题描述】：

我正在使用 play2 和 reactivemongo 从 mongodb 获取结果。结果的每一项都需要转换以添加一些元数据。之后我需要对其进行一些排序。

为了处理我使用enumerate()的转换步骤：

def ideasEnumerator = collection.find(query)
    .options(QueryOpts(skipN = page))
    .sort(Json.obj(sortField -> -1))
    .cursor[Idea]
    .enumerate()

然后我创建一个Iteratee如下：

val processIdeas: Iteratee[Idea, Unit] =
  Iteratee.foreach[Idea] { idea =>
    resolveCrossLinks(idea) flatMap { idea =>
      addMetaInfo(idea.copy(history = None))
    }
  }

最后我喂了 Iteratee：

ideasEnumerator(processIdeas)

现在我被困住了。我看到的每个示例都在foreach 中做了一些println，但似乎并不关心最终结果。

那么当所有文档都返回并转换后，我如何获得序列、列表或其他可以进一步处理的数据类型？

【问题讨论】：

Iteratee.foreach 纯粹是面向副作用的（它实际上是大多数集合类型上foreach 的约定）并返回Unit。你想要的可能是fold 的变体。
如果您可以使用List，您可以使用enum.run(Iteratee.getChunks)。

标签： scala playframework playframework-2.0 reactivemongo

【解决方案1】：

将 Iteratee 的签名从 Iteratee[Idea, Unit] 更改为 Iteratee[Idea, Seq[A]]，其中 A 是类型。基本上 Iteratee 的第一个参数是输入类型，第二个参数是输出类型。在您的情况下，您将输出类型指定为 Unit。

看看下面的代码。它可能无法编译，但它为您提供了基本用法。

ideasEnumerator.run(
  Iteratee.fold(List.empty[MyObject]) { (accumulator, next) => 
    accumulator + resolveCrossLinks(next) flatMap { next => 
      addMetaInfo(next.copy(history = None))
    } 
  }
) // returns Future[List[MyObject]]

如您所见，Iteratee 只是一个状态机。只需提取该 Iteratee 部分并将其分配给 val：

val iteratee = Iteratee.fold(List.empty[MyObject]) { (accumulator, next) => 
        accumulator + resolveCrossLinks(next) flatMap { next => 
          addMetaInfo(next.copy(history = None))
        } 
      }

并在任何需要将您的想法转换为 List[MyObject] 的地方随意使用它

【讨论】：

如果从 Iteratee 抛出异常，它不会传播到 Enumerator，因此 Future.map {}.recoverWith{} 中的后续 recoverWith 不起作用。我们如何解决这个问题？
嘿@Anand！错误确实会传播。刚刚尝试过 :) 这是一个要点：gist.github.com/vaedama/ec4019d50ef09dea1f554c3a856007d4

【解决方案2】：

在你的回答的帮助下，我最终得到了

val processIdeas: Iteratee[Idea, Future[Vector[Idea]]] = 
    Iteratee.fold(Future(Vector.empty[Idea])) { (accumulator: Future[Vector[Idea]], next:Idea) =>
  resolveCrossLinks(next) flatMap { next =>
    addMetaInfo(next.copy(history = None))
  } flatMap (ideaWithMeta => accumulator map (acc => acc :+ ideaWithMeta))
}


val ideas = collection.find(query)
  .options(QueryOpts(page, perPage))
  .sort(Json.obj(sortField -> -1))
  .cursor[Idea]
  .enumerate(perPage).run(processIdeas)

这稍后需要ideas.flatMap(identity) 来删除返回的 Future

与创建列表并随后对其进行迭代相比，所获得的性能可以忽略不计。

【讨论】：