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