gistfile1.scala

object FutureOps extends FutureOps
trait FutureOps {
  def takeFirstMatch[A](futures: Traversable[Future[A]], predicate: A => Boolean)(implicit ec:ExecutionContext): Future[Option[A]] = {
    if(futures.nonEmpty) {
      val promise = Promise[Option[A]]()
      val completedCount = new java.util.concurrent.atomic.AtomicInteger(0)
      val allDoneLatch = Promise[Unit]
      def maybeAllDone() {
        // Note: check and execute normally creates a race but only one of the futures can cause the final count to be reached
        if(completedCount.incrementAndGet() == futures.size) {
          allDoneLatch.success(())
        }
      }

      Future.sequence(futures) onComplete { case _ =>
        allDoneLatch.future onSuccess { case _ =>
          val exceptions = futures.map(_.value).flatten.filter(_.isFailure).map(_.failed.get)
          if(!promise.isCompleted) {
            // If there was at least one success that didn't pass the predicate
            if(exceptions.size < futures.size) {
              exceptions.foreach(ex => ec.reportFailure(ex))
              promise.success(None)
            } else {
              promise.failure(new RuntimeException(s"All Futures Failed - StackTraces: ${exceptions.toList.zipWithIndex.map { case (e, i) => s"Exception ${i} ${ExceptionUtils.getStackTrace(e)}"}}"))
            }
          } else {
            exceptions.foreach(ex => ec.reportFailure(ex))
          }
        }
      }

      futures foreach { x => x.onComplete {
        case Success(a) if predicate(a) =>
          promise.trySuccess(Some(a))
          maybeAllDone()
        case _ =>
          maybeAllDone()
      }

      }
      promise.future
    }
    else {
      Future.successful(None)
    }
  }

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