ReactiveMongo 与 Akka Streams 自定义源答案

【问题标题】：ReactiveMongo with Akka Streams Custom SourceReactiveMongo 与 Akka Streams 自定义源
【发布时间】：2017-11-27 18:31:06
【问题描述】：

我正在使用 reactivemongo-akka-stream 并尝试转换 AkkaStreamCursor.documentSource。我现在看到两个问题：

documentation 声明此操作返回 Source[T, NotUsed] 但 collection.find(query).cursor[BSONDocument].doccumentSource() 返回 Source[BSONDocument, Future[State]]。有没有办法避免 State 对象？

假设我使用Future[State]，我希望获得以下类的来源

case class Inner(foo: String, baz: Int)

case class Outer(bar: Inner)

//
implicit object InnerReader extends BSONDocumentReader[Inner]//defined

val getCollection: Future[BSONCollection] = connection.database("db").map(_.collection("things")

def stream()(implicit m: Materializer): Source[Outer, Future[State]] = {
 getCollection.map(_.find().cursor[Inner]().documentSource()).map(_.via(Flow[Inner].map(in => Outer(in))))

但不是返回一个我可以处理的Future[Source[Outer, Future[State]]，而是返回一个Future[Source[Inner, Future[State]]#Repr[Outer]]

bson 阅读器如何与这个库一起使用？

【问题讨论】：

链接文档中的第一个示例使用Future[State]。如果您不关心物化状态，请不要使用它。那么这不是 BSON 读者处理异步问题的责任。要展平 Future 的来源，您可以使用 flatMapConcat 和 fromFuture，或者从 Akka 2.5.1 Source.fromFutureGraph 开始。
BSON 处理异步问题是什么意思？我只想流入一个自定义案例类...我不太了解#Repr[Outer]的类型签名
您尝试过建议的flatMapConcat 或fromFutureGraph 解决方案吗？（这种Repr依赖类型在Akka Stream中很常见）。

标签： scala akka-stream reactivemongo

【解决方案1】：

根据 cchantep 的建议，我需要使用 fromFuture 和 flatMapConcat：

def stream()(implicit m: Materializer): Source[Outer, NotUsed] = {
 val foo = getCollection.map(x => col2Source(x))

 fix(foo).via(flowmap(map = Outer(_)))
}
def col2Source(col: BSONCollection): Source[Inner, Future[State]] = {
 val cursor: AkkaStreamCursor[Inner] = 
   col.find(BSONDocument.empty).cursor[Inner]()

 cursor.documentSource()
}

def flowmap[In, Out](
 map: (In) => Out
): Flow[In, Out, NotUsed] = Flow[In].map(e => map(e))

def fix[Out, Mat](futureSource: Future[Source[Out, Mat]]): Source[Out, NotUsed] = {
 Source.fromFuture(futureSource).flatMapConcat(identity)
}

【讨论】：