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

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()