traverse.scala

import scala.concurrent.Future
import scala.concurrent.duration._
import java.util.concurrent.Executors

implicit val ec = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(5))

val r = Future {
  val ids = (0 to 10)
  println(ids)
  ids.toSeq
}.flatMap {
  _.sliding(3, 3).foldLeft(Future.successful(Seq.empty[Int])) { (acc, ids) =>
    var p = util.Random.alphanumeric.take(3).mkString
    acc.flatMap { rs =>
      Future.traverse(ids) { id =>
        Future {
          Thread.sleep(3 * 1000)
          println(s"${Thread.currentThread.getId} - $p - $id")
          id * 10
        }
      }.map(rs ++ _)
    }
  }
}

Await.result(r, 15.seconds)

def traverse[A, B](in: Seq[A], window: Int)(fn: A => Future[B])(implicit ec: ExecutionContext): Future[Seq[B]] =
  in.sliding(window, window).foldLeft(Future.successful(Seq.empty[B])) { (acc, is) =>
    acc.flatMap { rs =>
      println("=========")
      Future.traverse(is)(fn).map(rs ++ _)
    }
  }

def run(p: Int) = Await.result(
traverse((0 to 10).toSeq, p) { i =>
  Future {
    Thread.sleep(3 * 1000)
    println(s"${Thread.currentThread.getId} - $i")
    i * 10
  }
},
Duration.Inf
)