MongoDB 通过模式匹配获取正确的引用

Question

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

  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

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

    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 数据库中插入一条记录：

    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方法中进行模式匹配来获取正确的引用？

Answer 1

您需要创建一个具有两个分支的流，这两个分支终止于不同的接收器。实现这一目标的方法之一是使用 divertTo 运算符。

MongoDB 通过模式匹配获取正确的引用

MongoDB take the right reference with pattern matching

scala

akka

akka-stream

alpakka