如何将嵌套的 Future 排除在选项之外？

Question

我无法解决def finalFuture(): Future[Option[(A1, Option[B1])]] 的问题。在下面的代码中，它返回一个嵌套的 Future 和 Option，并出现此错误：

Error:(36, 5) type mismatch;
 found   : scala.concurrent.Future[Option[scala.concurrent.Future[Option[(A1, Option[B1])]]]]
 required: scala.concurrent.Future[Option[(A1, Option[B1])]]
    res
    ^

代码：

import scala.concurrent._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

case class A1(val id: Int)

case class B1(val id: Int)

object NestedExample {

  def outerFuture(): Future[Option[A1]] = {
    Future {
      Some(A1(id = 0))
    }
  }

  def innerFuture(num: Int): Future[Option[B1]] = {
    Future {
      if (num < 10) Some(B1(0)) else None
    }
  }

  def finalFuture(): Future[Option[(A1, Option[B1])]] = {
    val f1 = outerFuture()
    val res = for (o1 <- f1) yield {
      o1.map { a =>
        val f2 = innerFuture(a.id)
        val q = f2.map { bOpt =>
          val w:Option[(A1, Option[B1])] = bOpt.map(x => (a, Some(x)))
          val e = w.orElse(Some((a, None)))
          e
        }
        q
      }
    }
    res
  }

  def main(args: Array[String]): Unit = {
    val ff = finalFuture()
    ff.onSuccess { case x => println(x) }
    ff.onFailure { case x => println(x) }
    Await.result(ff, 5 seconds)
  }
}

我该如何解决？

更新1

根据此处的答案，finalFuture1 和 finalFuture2 仍然无法处理一个案例：

import scala.concurrent._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

case class A1(val id: Int)

case class B1(val id: Int)

object NestedExample {

  def outerFuture(num: Int): Future[Option[A1]] = {
    Future {
      if (num <= 5) None
      else if (num <= 10) { // 6 to 10
        Some(A1(num))
      } else { // >= 11
        throw new RuntimeException("out of range")
      }
    }
  }

  def innerFuture(num: Int): Future[Option[B1]] = {
    Future {
      num match {
        case 6 => None
        case 7 => Some(B1(num * 2))
        case _ => throw new RuntimeException("neither 6 nor 7")
      }
    }
  }

  def finalFuture1(num: Int): Future[Option[(A1, Option[B1])]] = {
    val f1 = outerFuture(num)
    val res = f1 flatMap { aOpt =>
      val x = aOpt.map { a =>
        val f2 = innerFuture(a.id)
        val q = f2.map { bOpt =>
          val w: Option[(A1, Option[B1])] = bOpt.map(x => (a, Some(x)))
          val e = w.orElse(Some((a, None)))
          e
        }
        q
      }.getOrElse(Future.successful(None))
      x
    }
    res
  }

  def finalFuture2(num: Int): Future[Option[(A1, Option[B1])]] = {
    for (
      o1 <- outerFuture(num);
      o2 <- o1 match {
        case Some(a) => innerFuture(a.id).map(b => Some(a, b))
        case None => Future(None)
      }
    ) yield o2
  }

  def main(args: Array[String]): Unit = {
    for (n <- List(5, 6, 7, 8, 11)) {
      println(n)
      val ff = finalFuture1(n)
      ff.onSuccess { case x => println(x) }
      ff.onFailure { case x => println(x.getMessage) }
      Await.result(ff, 5 seconds)
    }
  }
}

当 num 为 8 时失败，即每当 innerFuture 失败时，它会导致在 monad 中无法正确捕获异常：

5
None
6
Some((A1(6),None))
7
8
Some((A1(7),Some(B1(14))))
neither 6 nor 7
Exception in thread "main" java.lang.RuntimeException: neither 6 nor 7
    at NestedExample$$anonfun$innerFuture$1.apply(NestedExample.scala:30)
    at NestedExample$$anonfun$innerFuture$1.apply(NestedExample.scala:27)
    at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
    at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
    at scala.concurrent.impl.ExecutionContextImpl$AdaptedForkJoinTask.exec(ExecutionContextImpl.scala:121)
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

对于finalFuture1(8)，结果应该是Some((A1(8), None))。

更新2

fallbackTo 一切顺利！

  def finalFuture1(num: Int): Future[Option[(A1, Option[B1])]] = {
    val f1 = outerFuture(num)
    val res = f1 flatMap { aOpt =>
      val x = aOpt.map { a =>
        val f2 = innerFuture(a.id)
        val q = f2.map { bOpt =>
          val w: Option[(A1, Option[B1])] = bOpt.map(x => (a, Some(x)))
          val e = w.orElse(Some((a, None)))
          e
        }
        q.fallbackTo(Future(Some((a, None))))
      }
      x.getOrElse(Future.successful(None))
    }
    res.fallbackTo(Future(None))
  }


  def finalFuture2(num: Int): Future[Option[(A1, Option[B1])]] = {
    val f1 = for {
      o1 <- outerFuture(num)
      o2 <- {
        o1 match {
          case Some(a) =>
            innerFuture(a.id)
              .map(b => Some(a, b))
              .fallbackTo(Future(Some((a, None))))
          case None => Future(None)
        }
      }
    } yield o2
    f1.fallbackTo(Future(None))
  }

但是我想知道它是否可以变得更漂亮？

Answer 1

也许这就是你想要做的：

def finalFuture(): Future[Option[(A1, Option[B1])]] = 
  for (
    o1 <- outerFuture();
    o2 <- o1 match {
      case Some(a) => innerFuture(a.id).map(b => Some(a, b))
      case _ => Future(None)
      }
  ) yield o2

Answer 2

我想这个应该可以工作，因为当你执行 Option.map(Future[]] if it is None 时，它​​的类型可能会发疯

这应该可以工作

import scala.concurrent._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

case class A1(val id: Int)

case class B1(val id: Int)

object NestedExample {

  def outerFuture(): Future[Option[A1]] = {
    Future {
      Some(A1(id = 0))
    }
  }

  def innerFuture(num: Int): Future[Option[B1]] = {
    Future {
      if (num < 10) Some(B1(0)) else None
    }
  }

  def finalFuture(): Future[Option[(A1, Option[B1])]] = {
    val f1 = outerFuture()

    val res = f1.flatMap(o1 => {
      o1.map { a =>
        val f2 = innerFuture(a.id)
        val q = f2.map { bOpt =>
          val w:Option[(A1, Option[B1])] = bOpt.map(x => (a, Some(x)))
          val e = w.orElse(Some((a, None)))
          e
        }
        q
      }.getOrElse(Future.successful(None))
    })

    res
  }

  def main(args: Array[String]): Unit = {
    val ff = finalFuture()
    ff.onSuccess { case x => println(x) }
    ff.onFailure { case x => println(x) }
    Await.result(ff, 5 seconds)
  }