Basic Future takeFirstMatch

gistfile1.scala

object FutureOps extends FutureOps
trait FutureOps {
  def takeFirstMatch[A](futures: Traversable[Future[A]], predicate: A => Boolean): Future[Option[A]] = {
    val promise = Promise[Option[A]]
    def aggregateFailures(futuresToFailureCheck: Traversable[Future[A]], exceptions: Seq[Throwable]): Unit = {
      futuresToFailureCheck match {
        case Nil => {
          Promise.failed(new RuntimeException(s"All Futures Failed - StackTraces: ${exceptions.zipWithIndex.map { case (e, i) => s"Exception ${i} ${ExceptionUtils.getStackTrace(e)}"}}"))
        }
        case head :: tail => head onFailure { case ex => aggregateFailures(tail, exceptions :+ ex)}
      }
    }

    futures.foreach(_.onSuccess { case a if predicate(a) => promise.trySuccess(Some(a))})

    aggregateFailures(futures,Nil)

    Future.sequence(futures) onComplete {
      case _ => {
        val completeSuccess = futures.map(_.value).map(_.get).filter(_.isSuccess)
        if (!completeSuccess.isEmpty) {
          completeSuccess.map(_.get).find(predicate) match {
            case Some(value) => promise.trySuccess(Some(value)) //just in case this executed first
            case _ => promise.trySuccess(None)
          }
        }
      }
    }
    
    promise.future
  }


  def takeFirst[A](futures: Traversable[Future[A]]) = takeFirstMatch(futures,{ A => true})
}