fork and join using Akka

Question

problem statement: I have a portfolio of securities that need to be processed in a parallel fashion. In Java i used a threadpool to process each security, and use a latch to countdown. Once complete I do some merging etc.

So I message my SecurityProcessor(which is an actor), and wait on all the futures to complete. In the end I use a MergeHelper to do the post-processing. The SecurityProcessor takes a security, does some i/o and processing and replies a Security

  val listOfFutures = new ListBuffer[Future[Security]]()
  var portfolioResponse: Portfolio = _
  for (security <- portfolio.getSecurities.toList) {
    val securityProcessor = actorOf[SecurityProcessor].start()
    listOfFutures += (securityProcessor ? security) map {
      _.asInstanceOf[Security]
    }
  }
  val futures = Future.sequence(listOfFutures.toList)
  futures.map {
    listOfSecurities =>
      portfolioResponse = MergeHelper.merge(portfolio, listOfSecurities)
  }.get

Is this design correct, and is there a better/cooler way to implement this common problem using akka?

Answer 1

val futureResult = Future.sequence(
                  portfolio.getSecurities.toList map { security => (actorOf[SecurityProcessor].start() ? security).mapTo[Security] }
                ) map { securities => MergeHelper.merge(portfolio, securities) }