Akka Streams aufgeteilt Strom nach Typ

Question

ich die folgenden einfachen Fall Klassenhierarchie haben:

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

Und ich habe ein Flow[Message, Message, NotUsed] (von einem Websocket-basiertes Protokoll mit Codec bereits vorhanden).

Ich möchte diese Flow[Message] in separate Flüsse für Foo und Baz Typen demultiplexen, da diese auf völlig verschiedenen Wegen verarbeitet werden.

Was ist der einfachste Weg? Sollte offensichtlich sein, aber ich vermisse etwas ...

Answer 1

Eine Möglichkeit ist die Erstellung eines RunnableGraph, das die Flüsse für jeden Nachrichtentyp enthält.

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