RunAll.scala

package org.x4444.onakka

import java.io.InputStream
import java.util.concurrent._

/**
  * RunAll
  */
object RunAll {

  def solve[T](e: Executor, solvers: List[Callable[T]]): List[T] = {
    val ecs = new ExecutorCompletionService[T](e)

    solvers.foreach(s => ecs.submit(s))

    // get results from All
    solvers.map { s =>
      ecs.take().get()
    }
  }
}

object Example {
  def main(args: Array[String]): Unit = {
    val tpe = Executors.newFixedThreadPool(5)

    val res = Arm(getClass.getResourceAsStream("/a.txt")).map { is =>
      val solvers = List(MyRead(is), MyRead(is), MyRead(is),
        MyRead(is), MyRead(is), MyRead(is), MyRead(is))
      
      RunAll.solve(tpe, solvers)
    }

    println(res.mkString(","))
  }
}

case class MyRead(is: InputStream) extends Callable[String] {
  override def call(): String = {
    val arr = Array.fill[Byte](1)(0)
    val l = is.read(arr)
    if (l == -1) null
    else new String(arr, "UTF-8")
  }
}