如何一次又一次地调用一个方法，直到它返回一个包含“None”的“Future”值

Question

给定一个返回 Future 这样的方法...

def remove(id: String): Future[Option[User]] = Future {
  // removes and returns the user identified by `id`
}

...如何一次又一次地调用它，直到它返回一个包含None 的Future 值？

编辑

也许值得一提的是，我不需要收集结果。只要找到要删除的用户，我只需要调用该方法。这个想法是有一个loop 当remove 返回Future[None] 时停止。

Answer 1

之前有人评论说没有意义。

令我惊讶的是，懒惰地消费期货并没有什么捷径可走。 Future.find 类似于 firstCompletedOf，并不代表 find first in traversable order。

scala> import concurrent._, ExecutionContext.Implicits._
import concurrent._
import ExecutionContext.Implicits._

scala> import java.util.concurrent.atomic._
import java.util.concurrent.atomic._

scala> val count = new AtomicInteger(10)
count: java.util.concurrent.atomic.AtomicInteger = 10

scala> def f(s: String) = Future { if (count.decrementAndGet <= 0) None else Some(s) }
f: (s: String)scala.concurrent.Future[Option[String]]

scala> def g(ss: List[String]): Future[List[String]] = f("hello") flatMap { case None => Future.successful(ss) case Some(s) => g(s :: ss) }
g: (ss: List[String])scala.concurrent.Future[List[String]]

scala> g(Nil)
res0: scala.concurrent.Future[List[String]] = scala.concurrent.impl.Promise$DefaultPromise@65a15628

scala> .value
res1: Option[scala.util.Try[List[String]]] = Some(Success(List(hello, hello, hello, hello, hello, hello, hello, hello, hello)))

说明不阻塞的实用性：

scala> :pa
// Entering paste mode (ctrl-D to finish)

import scala.util._
import concurrent._, ExecutionContext.Implicits._
import java.util.concurrent.atomic._

class Work {
  val count = new AtomicInteger(10)
  def f(s: String) = Future {
    if (count.decrementAndGet <= 0) None else Some(s)
  } andThen {
    case Success(Some(x)) => Console println s"Calculated $x"
    case Success(None)    => Console println "Done."
    case _                => Console println "Failed."
  }
}

// Exiting paste mode, now interpreting.

import scala.util._
import concurrent._
import ExecutionContext.Implicits._
import java.util.concurrent.atomic._
defined class Work

显示Stream 版本，直到消费线程通过阻塞等待时才会计算前缀：

scala> val work = new Work
work: Work = Work@1b45c0e

scala> Stream continually work.f("hello") takeWhile { x => Await.result(x, duration.Duration.Inf).nonEmpty }
Calculated hello
res0: scala.collection.immutable.Stream[scala.concurrent.Future[Option[String]]] = Stream(scala.concurrent.impl.Promise$DefaultPromise@66629f63, ?)

scala> .toList
Calculated hello
Calculated hello
Calculated hello
Calculated hello
Calculated hello
Calculated hello
Calculated hello
Calculated hello
Done.
res1: List[scala.concurrent.Future[Option[String]]] = List(scala.concurrent.impl.Promise$DefaultPromise@66629f63, scala.concurrent.impl.Promise$DefaultPromise@610db97e, scala.concurrent.impl.Promise$DefaultPromise@6f0628de, scala.concurrent.impl.Promise$DefaultPromise@3fabf088, scala.concurrent.impl.Promise$DefaultPromise@1e392345, scala.concurrent.impl.Promise$DefaultPromise@12f3afb5, scala.concurrent.impl.Promise$DefaultPromise@4ced35ed, scala.concurrent.impl.Promise$DefaultPromise@2c22a348, scala.concurrent.impl.Promise$DefaultPromise@7bd69e82)

scala> .foreach (Console println _.value.get)
Success(Some(hello))
Success(Some(hello))
[snip]

另一种行为，可能更可取，你会得到一个 Future 来保存计算前缀的结果：

scala> :pa
// Entering paste mode (ctrl-D to finish)

  val work = new Work
  def g(ss: List[String]): Future[List[String]] = work.f("hello") flatMap {
    case None => Future.successful(ss)
    case Some(s) => g(s :: ss)
  }

// Exiting paste mode, now interpreting.

work: Work = Work@796d3c9f
g: (ss: List[String])scala.concurrent.Future[List[String]]

scala> g(Nil)
Calculated hello
Calculated hello
res3: scala.concurrent.Future[List[String]] = scala.concurrent.impl.Promise$DefaultPromise@99a78d7
Calculated hello
Calculated hello
Calculated hello

scala> Calculated hello
Calculated hello
Calculated hello
Calculated hello
Done.

使用未来：

scala> .value
res5: Option[scala.util.Try[List[String]]] = Some(Success(List(hello, hello, hello, hello, hello, hello, hello, hello, hello)))

Answer 2

Stream#continually 无休止地做同样的事情，Stream#takeWhile 在某个点停止它。 http://www.scala-lang.org/api/2.11.0/index.html#scala.collection.immutable.Stream

Stream.continually(/*remove*/).takeWhile(/*not Future[None]*/)

Answer 3

这里是：

import concurrent._, ExecutionContext.Implicits._
import java.util.concurrent.atomic._

val count = new AtomicInteger(10)

def f(s: String) = Future {
  if (count.decrementAndGet <= 0) None else Some(s)
}

Iterator continually {
  f("hello")
} takeWhile {
  Await.result(_, duration.Duration.Inf).nonEmpty
} foreach { _.map { _.map {
  println
}}

希望对你有帮助。