an experiment in pushing the next Future that is ready in an enumerator to serve results as they are computed with Play Framework.

asSoonAsPossibleEnumeratee

  import scala.concurrent._
  import scala.concurrent.{ Future, Await }
  import scala.util.Random
  import play.api.libs.concurrent.Execution.Implicits.defaultContext
  import scala.util.{ Try, Success, Failure }
  import scala.concurrent.duration._
  import play.api.libs.iteratee._

  def takeOrDeadline[E](count: Int, deadline: Deadline): Enumeratee[E, E] = new Enumeratee.CheckDone[E, E] {

    def step[A](remaining: Int)(k: K[E, A]): K[E, Iteratee[E, A]] = {

      case in @ Input.El(_) if remaining > 0 && deadline.hasTimeLeft =>
        new Enumeratee.CheckDone[E, E] { def continue[A](k: K[E, A]) = Cont(step(remaining - 1)(k)) } &> k(in)

      case in @ Input.Empty if remaining > 0 && deadline.hasTimeLeft =>
        new Enumeratee.CheckDone[E, E] { def continue[A](k: K[E, A]) = Cont(step(remaining)(k)) } &> k(in)

      case in if deadline.hasTimeLeft => Done(Cont(k), in)
      case in =>
        Done(Cont(k), in)
    }

    def continue[A](k: K[E, A]) = Cont(step(count)(k))

  }                                          


    val doneMsg: Enumerator[Int] = Enumerator[Int](90000) //the final message
                                            

  val limitCutoff = takeOrDeadline[Int](
    5, 1 minute fromNow)                         
    
  val (enum, chan) = Concurrent.broadcast[Int]    
  
  val printIt = Iteratee.foreach[Int](m => println("done: " + m))
                                                 
  enum >>> doneMsg |>>> printIt              

  val futures = for { x <- 0 to 10 } yield {
    val f = future {
      Thread.sleep(Random.nextInt(30) * 10)
      if (x == 9) throw new Exception("I am not happy")
      x
    }
    f.andThen {
      case Success(x) =>
        chan.push(x)
        System.out.println(s"> ${x} is done")
        x
      case Failure(e) =>
        System.out.println("failed")
        throw e
    }
  }                                            

  val all = Future.sequence(futures).andThen {
    case Success(s) =>
      System.out.println("all done")
      chan.eofAndEnd()
      true
    case Failure(e) =>
      System.out.println("doen with some failures")
      chan.eofAndEnd()
      false
  }                                             
  try {
    Await.result(all, 1 second)
  } catch {
    case e: Exception => println("caught")
  }