How to poll with a Future in Scala?

Question

I want to poll an API endpoint until it reaches some condition. I expect it to reach this condition in couple of seconds to a minute. I have a method to call the endpoint that returns a Future. Is there some way I can chain Futures together to poll this endpoint every n milliseconds and give up after t tries?

Assume I have a function with the following signature:

def isComplete(): Future[Boolean] = ???

The simplest way to do this in my opinion would be to make everything blocking:

def untilComplete(): Unit = {
  for { _ <- 0 to 10 } {
    val status = Await.result(isComplete(), 1.seconds)
    if (status) return Unit
    Thread.sleep(100)
  }
  throw new Error("Max attempts")
}

But this may occupy all the threads and it is not asynchronous. I also considered doing it recursively:

def untilComplete(
    f: Future[Boolean] = Future.successful(false),
    attempts: Int = 10
  ): Future[Unit] = f flatMap { status =>
    if (status) Future.successful(Unit)
    else if (attempts == 0) throw new Error("Max attempts")
    else {
      Thread.sleep(100)
      untilComplete(isComplete(), attempts - 1)
    }
}

However, I am concerned about maxing out the call stack since this is not tail recursive.

Is there a better way of doing this?

Edit: I am using akka

Laciniate

For what you need, you might want to consider using Akka's scheduler. – Supplejack

Answer 1

You could use Akka Streams. For example, to call isComplete every 500 milliseconds until the result of the Future is true, up to a maximum of five times:

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{ Sink, Source }
import scala.concurrent.Future
import scala.concurrent.duration._

def isComplete(): Future[Boolean] = ???

implicit val system = ActorSystem("MyExample")
implicit val materializer = ActorMaterializer()
implicit val ec = system.dispatcher

val stream: Future[Option[Boolean]] =
  Source(1 to 5)
    .throttle(1, 500 millis)
    .mapAsync(parallelism = 1)(_ => isComplete())
    .takeWhile(_ == false, true)
    .runWith(Sink.lastOption)

stream onComplete { result =>
  println(s"Stream completed with result: $result")
  system.terminate()
}

Damiendamietta

This is a great answer. I actually need some data from isComplete that isn't represented in my simplified implementation and I got this data using inclusive = true on the takeWhile and Sink.lastOption. Leaving this hear so others can do this as well. – Laciniate

This doesn't ensure we don't try further if we get a successful result before making all the 5 tries, isn't it? – Eniwetok

Answer 2

It is actually not recursive at all, so the stack will be fine.

One improvement to your approach I can think of is to use some sort of scheduler instead of Thread.sleep so that you don't hold up the thread.

This example uses standard java's TimerTask, but if you are using some kind of a framework, like akka, play or whatever, it probably has its own scheduler, that would be a better alternative.

object Scheduler {
   val timer = new Timer(true)
   def after[T](d: Duration)(f :=> Future[T]): Future[T] = {
     val promise = Promise[T]()
     timer.schedule(TimerTask { def run() = promise.completeWith(f) }, d.toMillis)
     promise.future
   }
}


def untilComplete(attempts: Int = 10) = isComplete().flatMap { 
   case true => Future.successful(())
   case false if attempts > 1 => Scheduler.after(100 millis)(untilComplete(attempts-1))
   case _ => throw new Exception("Attempts exhausted.") 
}

Gerhardine

Thanks for the answer. I am using akka. Would the after from akka.pattern in place of your Scheduler.after work in this example? – Laciniate

Sure, that's perfect – Gerhardine

Also how is this not recursive? untilComplete calls itself – Laciniate

(untilComplete asks the Scheduler to call itself in a different Thread, with a different call stack.) – Draft

@Laciniate what Steve said :) It does not really call itself, it creates an anonymous function calling itself, that it then passes as an argument to Scheduler, that will at some point call it, after this call has completed. But more importantly, all of that happens inside an anonymous function passed to .flatMap, that itself will be invoked after the first call completes. That's the reason your original snippet isn't recursive either, even though it does not use the Scheduler – Gerhardine

Answer 3

I've given myself a library to do this. I have

trait Poller extends AutoCloseable {
  def addTask[T]( task : Poller.Task[T] ) : Future[T]
  def close() : Unit
}

where a Poller.Task looks like

class Task[T]( val label : String, val period : Duration, val pollFor : () => Option[T], val timeout : Duration = Duration.Inf )

The Poller polls every period until the pollFor method succeeds (yields a Some[T]) or the timeout is exceeded.

As a convenience, when I begin polling, I wrap this into a Poller.Task.withDeadline:

final case class withDeadline[T] ( task : Task[T], deadline : Long ) {
  def timedOut = deadline >= 0 && System.currentTimeMillis > deadline
}

which converts the (immutable, reusable) timeout Duration of the task into a per-poll-attempt deadline for timing out.

To do the polling efficiently, I use Java's ScheduledExecutorService:

def addTask[T]( task : Poller.Task[T] ) : Future[T] = {
  val promise = Promise[T]()
  scheduleTask( Poller.Task.withDeadline( task ), promise )
  promise.future
}

private def scheduleTask[T]( twd : Poller.Task.withDeadline[T], promise : Promise[T] ) : Unit = {
  if ( isClosed ) { 
    promise.failure( new Poller.ClosedException( this ) )
  } else {
    val task     = twd.task
    val deadline = twd.deadline

    val runnable = new Runnable {

      def run() : Unit = {
        try {
          if ( ! twd.timedOut ) {
            task.pollFor() match {
              case Some( value ) => promise.success( value )
              case None          => Abstract.this.scheduleTask( twd, promise )
            }
          } else {
            promise.failure( new Poller.TimeoutException( task.label, deadline ) )
          }
        }
        catch {
          case NonFatal( unexpected ) => promise.failure( unexpected )
        }
      }
    }

    val millis = task.period.toMillis
    ses.schedule( runnable, millis, TimeUnit.MILLISECONDS )
  }
}

It seems to work well, without requiring sleeping or blocking of individual Threads.

(Looking at the library, there's lots that could be done to make it clearer, easier to read, and the role of Poller.Task.withDeadline would be clarified by making the raw constructor for that class private. The deadline should always be computed from the task timeout, should not be an arbitrary free variable.)

This code comes from here (framework and trait) and here (implementation). (If you want to use it outright maven coordinates are here.)

Draft