0

我想处理 ea。异步查询获取（每个查询可能多次获取）。为了做到这一点，我将处理函数（返回 a Future）传递给我的查询方法，以便为 ea 调用它。拿来。我事先不知道查询的结果大小；我只知道我提取的最大尺寸。因此，我的查询返回一个Observable（而不是一个Listfor ex。我需要事先知道大小）。唯一的问题是，当我使用Observable createor时apply，它会在内部阻塞，直到 myFuture完成，然后再调用下一个 onNext ——实际上，消除了我希望从期货中获得的性能提升。Observable from工厂方法不会阻塞，但它需要一个Iterable. 我可以传递一个可变的Iterable并随着新的获取而增长。有人有一个更具参考性的透明sol'n？这是代码：

object Repository {
  def query(fetchSize: Int)(f: Set[Int] => Future[Set[Int]]): Observable[Future[Set[Int]]] = {
    // observable (as opposed to list) because modeling a process 
    // where the total result size is unknown beforehand. 
    // Also, not creating or applying because it blocks the futures
    val mut = scala.collection.mutable.Set[Future[Set[Int]]]()
    val obs = Observable.from(mut)
    1 to 2100 by fetchSize foreach { i =>
      mut += f(DataSource.fetch(i, fetchSize))
    }
    obs
  }
}
4

1 回答 1

0

我能够通过使用 foldLeft 来消除可变性：

(1 to 21 by fetchSize).foldLeft(Observable just Future((Set[Int]()))) { (obs, i) =>
  obs + f(DataSource.fetch(i)())
}

在哪里：

implicit class FutureObservable(obs: Observable[Future[Set[Int]]]) {
  def +(future: Future[Set[Int]]) =
  obs merge (Observable just future)
}

唯一的事情是我不喜欢我必须做的事情，创建一个Observable编译器没有抱怨的空。如果有人有更好的答案，请发布它，我会标记它。

于 2015-11-22T00:52:07.470 回答