title: How to create an akka-stream Source from a Flow that generate values recursively?
body: <p>I need to traverse an API that is shaped like a tree. For example, a directory structure or threads of discussion. It can be modeled via the following flow:</p>
  
  <pre><code>type ItemId = Int
  type Data = String
  case class Item(data: Data, kids: List[ItemId])
  
  def randomData(): Data = scala.util.Random.alphanumeric.take(2).mkString 
  
  // 0 =&gt; [1, 9]
  // 1 =&gt; [10, 19]
  // 2 =&gt; [20, 29]
  // ...
  // 9 =&gt; [90, 99]
  // _ =&gt; []
  // NB. I don't have access to this function, only the itemFlow.
  def nested(id: ItemId): List[ItemId] =
    if (id == 0) (1 to 9).toList
    else if (1 &lt;= id &amp;&amp; id &lt;= 9) ((id * 10) to ((id + 1) * 10 - 1)).toList
    else Nil
  
  val itemFlow: Flow[ItemId, Item, NotUsed] = 
    Flow.fromFunction(id =&gt; Item(randomData, nested(id)))
  </code></pre>
  
  <p>How can I traverse this data? I got the following working:</p>
  
  <pre><code>import akka.NotUsed
  import akka.actor.ActorSystem
  import akka.stream._
  import akka.stream.scaladsl._
  
  import scala.concurrent.Await
  import scala.concurrent.duration.Duration
  
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()
  
  val loop = 
    GraphDSL.create() { implicit b =&gt;
      import GraphDSL.Implicits._
  
      val source = b.add(Flow[Int])
      val merge  = b.add(Merge[Int](2))
      val fetch  = b.add(itemFlow) 
      val bcast  = b.add(Broadcast[Item](2))
  
      val kids   = b.add(Flow[Item].mapConcat(_.kids))
      val data   = b.add(Flow[Item].map(_.data))
  
      val buffer = Flow[Int].buffer(100, OverflowStrategy.dropHead)
  
      source ~&gt; merge ~&gt; fetch           ~&gt; bcast ~&gt; data
                merge &lt;~ buffer &lt;~ kids  &lt;~ bcast
  
      FlowShape(source.in, data.out)
    }
  
  val flow = Flow.fromGraph(loop)
  
  
  Await.result(
    Source.single(0).via(flow).runWith(Sink.foreach(println)),
    Duration.Inf
  )
  
  system.terminate()
  </code></pre>
  
  <p>However, since I'm using a flow with a buffer, the Stream will never complete.</p>
  
  <blockquote>
    <p>Completes when upstream completes and buffered elements have been drained</p>
  </blockquote>
  
  <p><a href="https://doc.akka.io/api/akka/current/akka/stream/scaladsl/Flow.html#buffer(size:Int,overflowStrategy:akka.stream.OverflowStrategy):FlowOps.this.Repr[Out]" rel="noreferrer">Flow.buffer</a></p>
  
  <p>I read the <a href="https://doc.akka.io/docs/akka/current/stream/stream-graphs.html" rel="noreferrer">Graph cycles, liveness, and deadlocks</a> section multiple times and I'm still struggling to find an answer.</p>
  
  <p>This would create a live lock:</p>
  
  <pre><code>import java.util.concurrent.atomic.AtomicInteger
  
  def unfold[S, E](seed: S, flow: Flow[S, E, NotUsed])(loop: E =&gt; List[S]): Source[E, NotUsed] = {
    // keep track of how many element flows, 
    val remaining = new AtomicInteger(1) // 1 = seed
  
    // should be &gt; max loop(x)
    val bufferSize = 10000
  
    val (ref, publisher) =
      Source.actorRef[S](bufferSize, OverflowStrategy.fail)
        .toMat(Sink.asPublisher(true))(Keep.both)
        .run()
  
    ref ! seed
  
    Source.fromPublisher(publisher)
      .via(flow)
      .map{x =&gt;
        loop(x).foreach{ c =&gt;
          remaining.incrementAndGet()
          ref ! c
        }
        x
      }
      .takeWhile(_ =&gt; remaining.decrementAndGet &gt; 0)
  }
  </code></pre>
  
  <p><em>EDIT:</em> I added a git repo to test your solution <a href="https://github.com/MasseGuillaume/source-unfold" rel="noreferrer">https://github.com/MasseGuillaume/source-unfold</a></p>