Akka Strumienie podzielony przez strumień typu

Question

mam następujący prosty przypadek hierarchię klasy:

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

i mam Flow[Message, Message, NotUsed] (od Protokół A websocket opartego o kodek już na miejscu).

Chcę zdemultipleksować ten Flow[Message] na osobne przepływy dla typów Foo i Baz, ponieważ są one przetwarzane przez zupełnie różne ścieżki.

Jaki jest najprostszy sposób to robić? Powinien być oczywisty, ale brakuje mi czegoś ...

Answer 1

Jednym ze sposobów jest użycie Utwórz RunnableGraph, który zawiera Flows dla każdego typu wiadomości.

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