How to throttle Futures with one-second delay with Akka

user_3994249 0
Viewed 471

I have list of URIs, each of which I want to request with a one-second delay in between. How can I do that?

val uris: List[String] = List()
// How to make these URIs resolve 1 second apart?
val responses: List[Future[Response]] = uris.map(httpRequest(_))
akka-stream akka-http akka scala
3 Answers

You could create an Akka Streams Source from the list of URIs, then throttle the conversion of each URI to a Future[Response]:

def httpRequest(uri: String): Future[Response] = ???

val uris: List[String] = ???

val responses: Future[Seq[Response]] =
  Source(uris)
    .throttle(1, 1 second)
    .mapAsync(parallelism = 1)(httpRequest)
    .runWith(Sink.seq[Response])
user_8654010

Something like this perahps:

  @tailrec
  def withDelay(
    uris: Seq[String], 
    delay: Duration = 1 second, 
    result: List[Future[Response]] = Nil,
  ): Seq[Future[Response]] = uris match {
     case Seq() => result.reversed
     case (head, tail@_*) => 
        val v = result.headOption.getOrElse(Future.successful(null))
          .flatMap { _ => 
            akka.pattern.after(delay, context.system.scheduler)(httpRequest(head))
          }
        withDelay(tail, delay, v :: result)
   }

this has a delay before the first execution as well, but I hope, it's clear enough how to get rid of it if necessary ... Another caveat is that this assumes that all futures succeed. As soon as one fails, all subsequent processing is aborted. If you need a different behavior, you may want to replace the .flatMap with .transform or add a .recover etc.

You can also write the same with .foldLeft if preferred: 

  uris.foldLeft(List.empty[Future[Response]]) { case (results, next) => 
    results.headOption.getOrElse(Future.successful(null))
      .flatMap { _ => 
        akka.pattern.after(delay, context.system.scheduler)(httpRequest(next))
      } :: results
  }.reversed
user_42545170
user_53099010
Related