Можно ли порождать потоки в Акка-Потоках на основе другого потока?

Question

В качестве примера BroadcastHub можно развернуть динамически рабочих, которые слушают одного и того же производителя. Но это нерестилирование должно выполняться явно в коде. Интересно, может ли он быть закодирован как реакция события в потоке.

В приведенном ниже примере я хотел бы создать еще 2 рабочих после получения сообщения «Спаун» в потоке spawns. Является ли это возможным?

package com.example 

import akka.actor.ActorSystem 
import akka.stream.ActorMaterializer 
import akka.stream.scaladsl.{BroadcastHub, Flow, Keep, Sink, Source} 

import scala.concurrent.duration._ 

object TestApp extends App { 
    implicit val system = ActorSystem("system") 
    implicit val materializer = ActorMaterializer() 

    val ticks = Source.tick(0.second, 1.second, "Tick").take(10) 

    val broadcaster = ticks.toMat(BroadcastHub.sink(bufferSize = 16))(Keep.right).run() 

    def prefixFlow(tag:String) = Flow[String].map(_ + s" from $tag").to(Sink.foreach(println)) 

    // Print out messages from the producer in two independent consumers 
    broadcaster.runWith(prefixFlow("1")) 
    broadcaster.runWith(prefixFlow("2")) 

    // Is it possible to spawn more flows based on another stream? 
    val spawns = Source.tick(2.second, 3.second, "Spawn").take(2) 
    // spawns.foreach(broadcaster.runWith(prefixFlow("XXX")) 
} 

Answer 1

В вашем конкретном примере вы могли бы найти простой map?

val spawns = Source 
    .tick(2.second, 3.second, "Spawn") 
    .take(2) 
    .map(_ ⇒ broadcaster.runWith(prefixFlow("XXX"))) 
    .runWith(Sink.ignore)