scala - Akka Streams 按类型拆分流

Question

我有以下简单的案例类层次结构：

sealed trait Message
case class Foo(bar: Int) extends Message
case class Baz(qux: String) extends Message

我有一个Flow[Message, Message, NotUsed]（来自基于 Websocket 的协议，编解码器已经到位）。

我想将其解复用Flow[Message]为 Foo 和 Baz 类型的单独流，因为它们是由完全不同的路径处理的。

最简单的方法是什么？应该很明显，但我错过了一些东西......

Answer 1

一种方法是使用创建一个包含每种消息类型的流的 RunnableGraph。

val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>

  val in = Source(...)  // Some message source
  val out = Sink.ignore

  val foo = builder.add(Flow[Message].map (x => x match { case f@Foo(_) => f }))
  val baz = builder.add(Flow[Message].map (x => x match { case b@Baz(_) => b }))
  val partition = builder.add(Partition[Message](2, {
    case Foo(_) => 0
    case Baz(_) => 1
  }))

  partition ~> foo ~> // other Flow[Foo] here ~> out
  partition ~> baz ~> // other Flow[Baz] here ~> out

  ClosedShape
}

g.run()