Elegante Art der Wiederverwendung akka-Strom fließt

Question

Ich bin auf der Suche nach einer Möglichkeit, akka-Stream-Ströme problemlos wieder zu verwenden.

Ich behandle die Flow I als Funktion wieder verwenden möchten, so würde ich gerne seine Unterschrift halten wie:

Flow[Input, Output, NotUsed]

Nun, wenn ich diesen Strom nutzen möchte ich in der Lage sein möchten ‚Call 'diesen Fluss und halten Sie das Ergebnis für die weitere Verarbeitung beiseite.

Also ich möchte mit Flow emiting [Input] beginnen, meine Strömung anwenden, und weiter mit Strömung emittieren [(Input, Output)].

Beispiel:

val s: Source[Int, NotUsed] = Source(1 to 10) 

val stringIfEven = Flow[Int].filter(_ % 2 == 0).map(_.toString) 

val via: Source[(Int, String), NotUsed] = ??? 

Nun ist dies auf einfache Art und Weise nicht möglich, da mit .via() Flow kombiniert ich nur [Output] emittierende

würde ein strömungs

val via: Source[String, NotUsed] = s.via(stringIfEven) 

Alternative ist meine wieder verwendbaren Strömungs [(Input, Output)] emittieren zu machen aber das erfordert, dass jeder Flow seinen Input durch alle Stufen treibt und meinen Code schlecht aussehen lässt.

So kam ich mit einem Kombinierer so zusammen:

def tupledFlow[In,Out](flow: Flow[In, Out, _]):Flow[In, (In,Out), NotUsed] = { 
    Flow.fromGraph(GraphDSL.create() { implicit b => 
    import GraphDSL.Implicits._ 

    val broadcast = b.add(Broadcast[In](2)) 
    val zip = b.add(Zip[In, Out]) 

    broadcast.out(0) ~> zip.in0 
    broadcast.out(1) ~> flow ~> zip.in1 

    FlowShape(broadcast.in, zip.out) 
}) 

}

, die den Eingang zu dem Fluss senden und auch in einer parallelen Linie direkt -> sowohl auf die ‚Zip‘ Stufe, in der ich Werte zu einem Tupel zusammenfüge. Es kann dann elegant angewendet werden:

val tupled: Source[(Int, String), NotUsed] = s.via(tupledFlow(stringIfEven)) 

Alles toll, aber wenn man ihnen Strömung einen ‚Filter‘ Betrieb zu tun - diese Kombinierer klemmt und hält weitere Ereignisse zu verarbeiten.

Ich denke, das ist wegen 'Zip' Verhalten, das alle Subflows dasselbe tun müssen - in meinem Fall wird ein gegebenes Objekt direkt von einem Zweig übergeben, so dass ein anderer Unterfluss dieses Element nicht ignorieren kann. filter(), und da es tut - der Fluss stoppt, weil Zip auf Push wartet.

Gibt es eine bessere Möglichkeit, die Fließzusammensetzung zu erreichen? Kann ich irgendetwas in meinem tupledFlow tun, um das gewünschte Verhalten zu erhalten, wenn 'flow' Elemente mit 'filter' ignoriert?

Answer 1

Zwei mögliche Ansätze - mit strittig Eleganz - sind:

1) vermeiden Filterstufen verwenden, die Filter in ein Flow[Int, Option[Int], NotUsed] mutiert. Auf diese Weise können Sie Ihren Zipping-Wrapper wie Ihr ursprünglicher Plan auf den gesamten Graphen anwenden. Der Code sieht jedoch etwas verdorbener aus, und es wird ein zusätzlicher Aufwand durch das Umgehen von None s verursacht.

2) Trennen Sie die Filter- und Transformationsstufen und wenden Sie die Filter vor dem Zipping-Wrapper an. Wahrscheinlich ein leichter und besserer Kompromiss.

val filterEven = Flow[Int].filter(_ % 2 == 0) 

val toString = Flow[Int].map(_.toString) 

val tupled: Source[(Int, String), NotUsed] = s.via(filterEven).via(tupledFlow(toString)) 

EDIT

3) für eine andere Lösung hier Posting Klarheit, nach den Diskussionen in den Kommentaren.

Dieser Flow-Wrapper ermöglicht das Emittieren jedes Elements aus einem bestimmten Fluss, gepaart mit dem ursprünglichen Eingabeelement, das es generiert hat. Es funktioniert für jede Art von innerem Fluss (emittiert 0, 1 oder mehr Elemente für jeden Eingang).

def tupledFlow[In,Out](flow: Flow[In, Out, _]): Flow[In, (In,Out), NotUsed] = 
    Flow[In].flatMapConcat(in => Source.single(in).via(flow).map(out => in -> out)) 

Answer 2

kam ich mit einer Implementierung von TupledFlow up, das, wenn eingewickelt arbeitet Flow verwendet filter() oder mapAsync() und wenn gewickelte Flow 0,1 oder N Elemente für jeden Eingang aussendet:

def tupledFlow[In,Out](flow: Flow[In, Out, _])(implicit materializer: Materializer, executionContext: ExecutionContext):Flow[In, (In,Out), NotUsed] = { 
    val v:Flow[In, Seq[(In, Out)], NotUsed] = Flow[In].mapAsync(4) { in: In => 
    val outFuture: Future[Seq[Out]] = Source.single(in).via(flow).runWith(Sink.seq) 
    val bothFuture: Future[Seq[(In,Out)]] = outFuture.map(seqOfOut => seqOfOut.map((in,_))) 
    bothFuture 
    } 
    val onlyDefined: Flow[In, (In, Out), NotUsed] = v.mapConcat[(In, Out)](seq => seq.to[scala.collection.immutable.Iterable]) 
    onlyDefined 
} 

den einzigen Nachteil ich Sehen Sie hier, dass ich einen Fluss für eine einzelne Entität instanziiere und materialisiere - nur um eine Vorstellung davon zu bekommen, einen Fluss als eine Funktion zu bezeichnen.

Ich habe keine Leistungstests auf diese - aber seit Heavy-Lifting in einem eingewickelten Flow durchgeführt wird, die in einer Zukunft ausgeführt wird - ich glaube, das wird in Ordnung sein.

Diese Implementierung übergibt alle Tests von https://gist.github.com/kretes/8d5f2925de55b2a274148b69f79e55ac#file-tupledflowspec-scala