仅调用 onComplete 时映射 scala rx observables

Question

我使用 scala observables 从 couchbase 获取项目，然后我使用 map、flatMap、zip 来转换结果。问题是，如果一个项目在 couchbase 中不存在，那么例如 .zip 不会仅调用 onComplete。示例：

import rx.lang.scala._

def getIdsWithValues(ids: Seq[String]): Map[K, V] = {
  val values = Observable.from(keyValueIds).flatMap(id => couchbaseBucket.async().get(id))
  values.zip(Observable.from(ids)) // zip is not called if no row in couchbase with id.
  ...
}

所以我想要：

1. 返回 k -> v
的映射
2. 我让 .zip 将 k 耦合到返回的 v（如果不存在，我希望 v 类似于 None。
3. 我看到如果 db 中不存在任何项目，则根本不调用 zip。

我想到在运行上面的代码后，扫描ids 输入参数，并为每个没有压缩值的参数添加一个 id 到它的值，但这就像添加另一个流，我希望 zip 处理这两个现有和不存在的行。

我应该如何处理？如何让.zip 处理现有行和非现有行？

Answer 1

不要使用zip() 运算符。相反，只需使用flatMap() 和materialize().take(1)。 materialize() 会将onComplete() 事件转换为可以映射到None 的Notification，而带有值的Notification 将映射到Some(value)。

def getIdsWithValues(ids: Seq[String]): Map[K, V] = {
  val values = Observable.from(keyValueIds)
   .flatMap(id => couchbaseBucket.async()
     .get(id)
     .materialize()
     .take(1)
     .map( res => if ( res.isOnComplete() )
                     (id, None)
                  else 
                     (id, Some(res.getValue))
  ...
}