Waiting for recursive scala futures

Question

simple code example that describes my problem:

import scala.util._ import scala.concurrent._ import scala.concurrent.duration._ import ExecutionContext.Implicits.global class LoserException(msg: String, dice: Int) extends Exception(msg) { def diceRoll: Int = dice } def aPlayThatMayFail: Future[Int] = { Thread.sleep(1000) //throwing a dice takes some time... //throw a dice: (1 + Random.nextInt(6)) match { case 6 => Future.successful(6) //I win! case i: Int => Future.failed(new LoserException("I did not get 6...", i)) } } def win(prefix: String): String = { val futureGameLog = aPlayThatMayFail futureGameLog.onComplete(t => t match { case Success(diceRoll) => "%s, and finally, I won! I rolled %d !!!".format(prefix, diceRoll) case Failure(e) => e match { case ex: LoserException => win("%s, and then i got %d".format(prefix, ex.diceRoll)) case _: Throwable => "%s, and then somebody cheated!!!".format(prefix) } }) "I want to do something like futureGameLog.waitForRecursiveResult, using Await.result or something like that..." } win("I started playing the dice") 

This simple example illustrates what I want to do. in principle, if I put it in words, I want to wait for the result for some calculations, when I compose different actions on the previous success or unsuccessful attachments.

so how would you implement the win method?

my "real world" problem, if it matters, uses dispatch for asynchronous HTTP calls, where I want to keep making http calls whenever the previous one ends, but the actions are different from the previous previous http call or not.

Answer 1

You can recover a failed future with a recursive call:

 def foo(x: Int) = x match { case 10 => Future.successful(x) case _ => Future.failed[Int](new Exception) } def bar(x: Int): Future[Int] = { foo(x) recoverWith { case _ => bar(x+1) } } scala> bar(0) res0: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@64d6601 scala> res0.value res1: Option[scala.util.Try[Int]] = Some(Success(10)) 

recoverWith takes a PartialFunction[Throwable,scala.concurrent.Future[A]] and returns a Future[A] . You have to be careful because it will use quite some memory when there will be many recursive calls here.

Answer 2

Since drexin answered the part about handling and recovering exceptions, let me try to answer the question about the recursive function related to futures. I believe that using Promise will help you achieve your goal. The modified code will look like this:

 def win(prefix: String): String = { val prom = Promise[String]() def doWin(p:String) { val futureGameLog = aPlayThatMayFail futureGameLog.onComplete(t => t match { case Success(diceRoll) => prom.success("%s, and finally, I won! I rolled %d !!!".format(prefix, diceRoll)) case Failure(e) => e match { case ex: LoserException => doWin("%s, and then i got %d".format(prefix, ex.diceRoll)) case other => prom.failure(new Exception("%s, and then somebody cheated!!!".format(prefix))) } }) } doWin(prefix) Await.result(prom.future, someTimeout) } 

Now this will not be true recursion in the sense that it will grow one long stack due to the fact that futures are asynchronous, but this is similar to recursion in spirit. Using a promise gives you something to block while recursion does this, blocking the caller from what happens behind the scenes.

Now, if I were to do this, I would suggest overriding things like this:

 def win(prefix: String): Future[String] = { val prom = Promise[String]() def doWin(p:String) { val futureGameLog = aPlayThatMayFail futureGameLog.onComplete(t => t match { case Success(diceRoll) => prom.success("%s, and finally, I won! I rolled %d !!!".format(prefix, diceRoll)) case Failure(e) => e match { case ex: LoserException => doWin("%s, and then i got %d".format(prefix, ex.diceRoll)) case other => prom.failure(new Exception("%s, and then somebody cheated!!!".format(prefix))) } }) } doWin(prefix) prom.future } 

So you can defer the decision about whether to block or use asynchronous callbacks to the caller of this function. This is more flexible, but also provides the caller with the fact that you are performing async calculations, and I'm not sure if this will be acceptable for your scenario. I will leave this decision to you.

Answer 3

This works for me:

 def retryWithFuture[T](f: => Future[T],retries:Int, delay:FiniteDuration) (implicit ec: ExecutionContext, s: Scheduler): Future[T] ={ f.recoverWith { case _ if retries > 0 => after[T](delay,s)(retryWithFuture[T]( f , retries - 1 , delay)) } }