0

我正在尝试使用https://doc.akka.io/docs/alpakka/current/mongodb.html api 将记录取决于数据类型插入 mongodb。
我们先来看看数据类型：

  sealed trait MsgDoc {
    }

    final case class MsgPreFailure(raw: String, reasons: Chain[String]) extends MsgDoc

    final case class MsgProceed(raw: String, status: MsgStatus) extends MsgDoc


    sealed trait MsgStatus {

    }

    case object MsgSuccess extends MsgStatus

    final case class MsgFailure(reasons: Chain[String]) extends MsgStatus

    final case class MsgUnknown(reason: String) extends MsgStatus

对于MsgDocsum 类型的每个居民，我将创建他们自己的注册表，因为我想存储在不同的集合中

    val preFailureRegistry = fromRegistries(fromProviders(classOf[MsgPreFailure]), DEFAULT_CODEC_REGISTRY)
    val proceedRegistry = fromRegistries(fromProviders(classOf[MsgProceed]), DEFAULT_CODEC_REGISTRY)

    private val client = MongoClients.create("mongodb://localhost:27017")
    private val db = client.getDatabase("Message")

    private val preFailureColl = db
      .getCollection("Failure", classOf[MsgPreFailure])
      .withCodecRegistry(preFailureRegistry)


    private val proceedColl = db
      .getCollection("Proceed", classOf[MsgProceed])
      .withCodecRegistry(proceedRegistry)

然后让我们尝试在 mongo db 中插入一条记录：

    Source
      .single[MsgDoc](MsgPreFailure("Test", Chain("Foo", "Foo", "Foo")))
      .runWith(MongoSink.insertOne(???))

作为该insertOne方法的参数，我想传递preFailureColl或proceedColl取决于数据类型（MsgPreFailure或MsgProceed）。在上面的示例中，它是MsgPreFailure，那么它应该执行以下调用

    Source
      .single[MsgDoc](MsgPreFailure("Test", Chain("Foo", "Foo", "Foo")))
      .runWith(MongoSink.insertOne(preFailureColl))

问题是，如何在insertOne方法内进行模式匹配以获取正确的引用？

4

1 回答 1

2

您需要创建一个具有两个分支的流，这些分支以不同的接收器终止。实现这一目标的方法之一是使用divertTo操作员。

于 2019-05-07T12:59:15.223 回答