akka/akka-core

Docs: Streams: Feedback loop in "Composing complex systems"

Ouverte

#20 563 ouverte le 19 mai 2016

 (2 commentaires) (0 réaction) (0 personne assignée)Scala (3 547 forks)batch import
1 - triagedhelp wantedt:docst:stream

Métriques du dépôt

Stars
 (13 277 étoiles)
Métriques de merge PR
 (Merge moyen 8j 19h) (10 PRs mergées en 30 j)

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

Guide contributeur