Akka flux partiels flux par type

Question

je la hiérarchie de classe simple cas suivant:

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

Et j'ai un Flow[Message, Message, NotUsed] (à partir d'un protocole basé Websocket avec codec déjà en place).

Je souhaite démultiplexer ce Flow[Message] en flux séparés pour les types Foo et Baz, car ceux-ci sont traités par des chemins complètement différents.

Quelle est la manière la plus simple de le faire? Devrait être évident, mais il me manque quelque chose ...

Answer 1

Une façon est d'utiliser créer un RunnableGraph qui inclut les flux pour chaque type de message.

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