Scala Synchronising Asynchronous calls with Future

问题:

I have a method that does a couple of database look up and performs some logic.

The MyType object that I return from the method is as follows:

case class MyResultType(typeId: Long, type1: Seq[Type1], type2: Seq[Type2])

The method definition is like this:

def myMethod(typeId: Long, timeInterval: Interval) = async {

  // 1. check if I can find an entity in the database for typeId  
  val myTypeOption = await(db.run(findMyTypeById(typeId))) // I'm getting the headOption on this result

  if (myTypeOption.isDefined) {

    val anotherDbLookUp = await(doSomeDBStuff) // Line A

    // the interval gets split and assume that I get a List of thse intervals
    val intervalList = splitInterval(interval)

    // for each of the interval in the intervalList, I do database look up
    val results: Seq[(Future[Seq[Type1], Future[Seq[Type2])] = for {
      interval <- intervalList
    } yield {
      (getType1Entries(interval), getType2Entries(interval))
    }
    // best way to work with the results so that I can return MyResultType 
  }
  else {
    None
  }
}

Now the getType1Entries(interval) and getType2Entries(interval) each returns a Future of Seq(Type1) and Seq(Type2) entries!

My problem now is to get the Seq(Type1) and Seq(Type2) out of the Future and stuff that into the MyResultType case class?

回答1:

You could refer to this question you asked

Scala transforming a Seq with Future

so you get the

val results2: Future[Seq([Iterable[Type1], [Iterable[Type2])] = ???

and then call await on it and you have no Futures at all, you can do what you want.

I hope I understood the question correctly.

Oh and by the way you should map myTypeOption instead of checking if it's defined and returning None if it's not

if (myTypeOption.isDefined) {
  Some(x)
} else {
  None
}

can be simply replaced with

myTypeOption.map { _ => // ignoring what actually was inside option
  x                     // return whatever you want, without wrapping it in Some
}

回答2:

If I understood your question correctly, then this should do the trick.

def myMethod(typeId: Long, timeInterval: Interval): Option[Seq[MyResultType]] = async {

    // 1. check if I can find an entity in the database for typeId
    val myTypeOption = await(db.run(findMyTypeById(typeId))) // I'm getting the headOption on this result

    if (myTypeOption.isDefined) {

      // the interval gets split and assume that I get a List of thse intervals
      val intervalList = splitInterval(interval)

      // for each of the interval in the intervalList, I do database look up
      val results: Seq[(Future[Seq[Type1]], Future[Seq[Type2]])] = for {
        interval <- intervalList
      } yield {
          (getType1Entries(interval), getType2Entries(interval))
      }
      // best way to work with the results so that I can return MyResultType
      Some(
        await(
          Future.sequence(
            results.map{
              case (l, r) =>
                l.zip(r).map{
                  case (vl, vr) => MyResultType(typeId, vl, vr)
                }
            })))
    }
    else {
      None
    }
  }

回答3:

There are two parts to your problem, 1) how to deal with two dependent futures, and 2) how to extract the resulting values.

When dealing with dependent futures, I normally compose them together:

val future1 = Future { 10 }
val future2 = Future { 20 }

// results in a new future with type (Int, Int)
val combined = for {
  a <- future1
  b <- future2
} yield (a, b)

// then you can use foreach/map, Await, or onComplete to do
// something when your results are ready..
combined.foreach { ((a, b)) =>
  // do something with the result here
}

To extract the results I generally use Await if I need to make a synchronous response, use _.onComplete() if I need to deal with potential failure, and use _.foreach()/_.map() for most other circumstances.