akka/akka-core
View on GitHubDocs: Streams: Feedback loop in "Composing complex systems"
Open
#20,563 opened on May 19, 2016
1 - triagedhelp wantedt:docst:stream
Repository metrics
- Stars
- (13,277 stars)
- PR merge metrics
- (Avg merge 8d 19h) (10 merged PRs in 30d)
Description
In "Composing Complex Systems", there are code examples which lead into a feedback loop. Since C and F are Merge Modules, C always sends an element to F, which always sends this element to C, and so forth. So, B can't Broadcast more elements and this stream hangs.
Example Code:
import akka.actor.ActorSystem
import akka.stream.scaladsl._
import akka.stream.{ActorMaterializer, FlowShape, Graph}
import akka.{Done, NotUsed}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
object StreamProgram2 {
import GraphDSL.Implicits._
private val graph: Graph[FlowShape[Int, Int], NotUsed] = GraphDSL.create() { implicit builder =>
val B = builder.add(Broadcast[Int](2))
val C = builder.add(Merge[Int](2))
val D = Flow[Int].map(_ + 1)
val E = builder.add(Balance[Int](2, waitForAllDownstreams = true))
val F = builder.add(Merge[Int](2))
//@formatter:off
C <~ F
B ~> C ~> F
B ~> D.map{x => println(s"D$x"); x} ~> E ~> F
//@formatter:on
FlowShape(B.in, E.out(1))
}.named("partial")
def main(args: Array[String]) {
implicit lazy val system = ActorSystem("example")
implicit val materializer = ActorMaterializer()
val sink: Sink[Any, Future[Done]] = Sink.foreach(println)
val source: Source[Int, NotUsed] = Source(1 to 999)
source.via(graph).runWith(sink).onComplete(_ => System.exit(0))
}
}