Akka Streams erstellen neue Quelle aus dem Ergebnis der vorherigen Quelle in einem Diagramm

Question

Ich versuche zu erstellen Graph, der 1 Eingabequelle nimmt, nimmt das letzte Element und produziert neue Quelle. Momentan habe ich das nur geschafft, indem ich den ersten Stream mit Sink.last materialisiert habe.

Source.fromFuture(curStream.runWith(Sink.last).map(last => createStreamFromOffset(last)))

Answer 1

die Lösung gefunden:

val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder => 
    import GraphDSL.Implicits._ 

    val in = Source(1 to 10) 

    val out = Sink.foreach(println) 

    val concat = builder.add(Concat[Int](2)) 

    val f1 = Flow[Int].fold(0) { (acc, el) => 
     el 
    } 

    in ~> f1 ~> concat 

    // creating new source from old stream last value 
    in.via(f1).flatMapConcat(_ => Source(666 to 670)) ~> concat 

    concat ~> out 

    ClosedShape 
    }) 
    g.run()