Composition of Conditional Futures in Scala

Question

I have some code that works for a simple case (2 Futures) but I can't quite find the way to generalize it for an unlimited number of Futures.

What I want to do is create some code that calls a future and, when that future is completed, call another one, and when this one is completed call another one, and so on.

I need the result of each call to be completed before calling the next one because I may not need to call it again (this is my stopping condition).

I know this can be solved explicitly with recursion, but I would like, if at all possible, a solution using for comprehensions and/or folds. I feel there must be a solution like that but I can't write it correctly.

Here's a a function that generates a a list of two random ints

def nextValue: Future[List[Int]] = Future{
  Thread.sleep(1000)
  val num1 = Random.nextInt(10)
  val num2 = Random.nextInt(10)
  List(num1,num2)
}

Now I would like to compose infinitely many such futures and join them all at the end (a single future of a list)

I'm just calling await.result for testing purposes

This works for 2 levels, but how to generalize it for N calls?

Await.result({
  nextValue.flatMap{ value1 =>
    nextValue.map{ value2 =>
      value1 ++ value2
    }
  }
},1.minute)

If each `Future` is launched only after the previous completes, what's the point of using `Future` at all? Could you wrap a `fold` or `Stream` in a single `Future` which will complete when the stopping condition has been met? — jwvh, Jul 01 '16 at 01:07

score 1 · Accepted Answer · edited May 23 '17 at 12:22

Future.sequence((0 to 100).map(_ => nextValue)).map(_.flatten)

Usage:

scala> Future.sequence((0 to 100).map(_ => nextValue)).map(_.flatten)
res3: scala.concurrent.Future[scala.collection.immutable.IndexedSeq[Int]] = scala.concurrent.impl.Promise$DefaultPromise@692e028d

scala> Await.result(res3, duration.Duration.Inf)
res4: scala.collection.immutable.IndexedSeq[Int] = Vector(5, 4, 3, 0, 4, 6, 0, 8, 0, 0, 4, 6, 2, 7, 4, 9, 8, 8, 6, 9, 1, 4, 5, 5, 8, 2, 2, 7, 6, 0, 5, 6, 6, 5, 9, 6, 3, 5, 7, 1, 3, 2, 5, 3, 3, 1, 8, 4, 6, 7, 5, 1, 3, 5, 7, 4, 1, 5, 9, 4, 5, 0, 1, 8, 5, 0, 0, 7, 4, 2, 4, 2, 2, 0, 4, 1, 6, 3, 8, 2, 1, 3, 5, 5, 8, 3, 6, 1, 3, 2, 9, 4, 9, 4, 7, 5, 7, 8, 7, 9, 5, 2, 5, 0, 2, 5, 6, 8, 6, 2, 3, 2, 0, 8, 9, 3, 9, 2, 7, 5, 1, 7, 1, 1, 8, 6, 8, 0, 5, 5, 6, 0, 8, 8, 3, 6, 4, 2, 7, 1, 0, 3, 3, 3, 3, 2, 8, 7, 3, 3, 5, 1, 6, 3, 3, 7, 8, 9, 9, 9, 1, 9, 9, 8, 1, 1, 5, 8, 1, 1, 7, 6, 3, 2, 5, 0, 4, 3, 0, 9, 9, 1, 2, 0, 3, 6, 2, 6, 8, 6, 6, 3, 9, 7, 1, 3, 5, 9, 6, 5, 6, 2)

Or with scalaz/cats:

//import scalaz._,Scalaz._
// --or--
//import cats.syntax.traverse._
//import cats.std.list._
//import cats.std.future._

(0 to 100).toList.traverseM(_ => nextValue)

Eplanation from here:

traverseM(f) is equivalent to traverse(f).map(_.join), where join is the scalaz name for flatten. It's useful as a kind of "lifting flatMap":

If you want some condition and still need to stay async, you can use fs2:

import fs2._
import fs2.util._

def nextValue: Task[List[Int]] = Task.delay{
  import scala.util.Random
  val num1 = Random.nextInt(10)
  val num2 = Random.nextInt(10)
  if(num1 > 5) List(num1,num2) else List()
}
Stream.repeatEval(nextValue).takeWhile(_.size > 0).runLog.map(_.flatten).unsafeRun

https://github.com/functional-streams-for-scala/fs2/blob/series/0.9/docs/guide.md

The same can be achieved with Iteratees:

Cats: https://github.com/travisbrown/iteratee or scalaz-iteratee package

Generally speaking, you can't implement this with fold because it's actually unfold and there is no good support for unfold in scala as standard library's Stream can't generalize over a Monad/ApplicativeFunctor (like EnumeratorT does) - you can only check a condition by doing Await.result on every unfolding step.

I thought the OP was looking for a collection of length determined by an exit condition, i.e. get `nextValue` until we get the terminus result. — jwvh, Jul 01 '16 at 01:31
you're right, I didn't read the question carefully - will update my answer — dk14, Jul 01 '16 at 01:34
yeah... I would like to test the result of getNext to see if it's valid and terminate if it isn't. Is this possible? — Felipe, Jul 01 '16 at 01:40
I've added solution with external library, now will add classic solution — dk14, Jul 01 '16 at 01:42
unfortunately I couldn't think of any implementation using unfold in standard library without `Await.result` on every step. I've added an explanation why. — dk14, Jul 01 '16 at 02:47

Composition of Conditional Futures in Scala

1 Answers1