Scatter-Gather with Akka Dataflow

Aggregator.scala

class Aggregator(recipients: Iterable[ActorRef]) extends Actor{
    
  def receive = {
    case msg @ Message(text) => 
      println("Started processing message `%s`" format(text))
      
      val result = Promise[String]()          
      val promises = List.fill(recipients.size)(Promise[String]())
      
      recipients.zip(promises).map{case (recipient, promise) =>
          (recipient !!! msg).map{result: String => 
            println("Binding recipient's response: %s" format(result))
            flow{        
              promise << result
            } 
          }
      }
      
      flow{        
        def gather(promises: List[CompletableFuture[String]], result: String = ""): String @cps[Future[Any]] = 
          promises match {
            case head :: tail => gather(tail, head() + result)
            case Nil => result  
          }
            
        println("Binding result...")
        result << gather(promises)
      }
      
      self.reply(result)
  }

}

AggregatorSnippet

class Aggregator(recipients: Iterable[ActorRef]) extends Actor

AggregatorSnippet_2.scala

    case msg @ Message(text) => 

      val promises = List.fill(recipients.size)(Promise[String]())
      
      recipients.zip(promises).map{case (recipient, promise) =>
          (recipient !!! msg).map{result: String => 
            println("Binding recipient's response: %s" format(result))
            flow{        
              promise << result
            } 
          }
      }

AggregatorSnippet_3.scala

      flow{        
        def gather(promises: List[CompletableFuture[String]], result: String = ""): String @cps[Future[Any]] = 
          promises match {
            case head :: tail => gather(tail, head() + result)
            case Nil => result  
          }
            
        println("Binding result...")
        result << gather(promises)
      }

Recipient.scala

class Recipient(id: Int) extends Actor {
  
  def receive = {
    case Message(msg) => 
      Thread.sleep(1000)
      self.reply("%s, [%s]! ".format(msg, id))
  }
  
}

Test.scala

    val recipients = (1 to 5).map(i => actorOf(new Recipient(i)).start)
    val aggregator = actorOf(new Aggregator(recipients)).start
    val results1 = aggregator !! Message("Hello")
    val results2 = aggregator !! Message("world")
    results1.map{ res => 
      println("Result: %s" format(res.asInstanceOf[Future[String]].get))
    }
    results2.map{ res => 
      println("Result: %s" format(res.asInstanceOf[Future[String]].get))
    }  

TestResults.sh

Started processing message `Hello`
Binding result...
Started processing message `world`
Binding result...
Binding recipient response: Hello, [3]!
Binding recipient response: Hello, [1]!
Binding recipient response: Hello, [4]!
Binding recipient response: Hello, [2]!
Binding recipient response: Hello, [5]!
Result: Hello, [5]! Hello, [4]! Hello, [3]! Hello, [2]! Hello, [1]!
Binding recipient response: world, [4]!
Binding recipient response: world, [3]!
Binding recipient response: world, [5]!
Binding recipient response: world, [2]!
Binding recipient response: world, [1]!
Result: world, [5]! world, [4]! world, [3]! world, [2]! world, [1]!