Akka 스트림 형

Question

에 의해 스트림을 분할 나는 다음과 같은 간단한 경우 클래스 계층이 있습니다

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

을 그리고 나는 Flow[Message, Message, NotUsed] (코덱과 웹 소켓 기반 프로토콜에서 이미 자리에)있다.

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