可缩放的可伸缩标的统一可观测的序列，而无需中间数据结构更新

Question

我有一个代码，该代码调用ouchbase来获取一些行，如下所示:

val gotValues: Observable[JsonDocument] = Observable.from(rowKeys).flatMap(id =>
      couchbaseBucket.async().get(id))

如果我有1,2,3,4,5,6作为输入行键，而DB中仅存在行1,2,3，则可观察对象将仅得到有关1,2,3的通知。

但是，我的要求是，我返回一个具有1,2,3 true(存在于db中)和4,5,6具有false(意味着不存在于DB中)的映射。我设法通过可观察到的scala做到了这一点，但是我正在使用中间 map 数据结构来返回包含所有ID的总 map 。以下是模拟我的问题的示例代码。

object Main extends App {
  import rx.lang.scala.Observable

  val idsToFetch = Seq(1,2,3,4,5,6)

  println(isInDBOrNot()) // {1=true, 2=true, 3=true, 4=false, 5=false, 6=false}

  private def isInDBOrNot(): ConcurrentHashMap[Int, Boolean] = {
    val inAndNotInDB = new java.util.concurrent.ConcurrentHashMap[Int, Boolean]
    // - How can I avoid the additional data structure?
    // - In this case a map, so that the function will return
    //   a map with all numbers and for each if exist in DB?
    // - I mean I want the function to return a map I don't 
    //   want to populate that map inside the observer,
    //   it's like a mini side effect I would rather simply 
    //   manipulate the stream.

    Observable.from(idsToFetch)
      .filterNot(x => x == 4 || x == 5 || x == 6) // Simulate fetch from DB, 4,5,6 do not exist in DB, so not returned.
      .subscribe(
      x => inAndNotInDB.put(x, true),
      e => println(e),
      () => idsToFetch.filterNot(inAndNotInDB.containsKey)
        .foreach(inAndNotInDB.put(_, false)) // mark all non-found as false.
    )

    inAndNotInDB
  }

}

无论如何，要在没有中间映射的情况下做到这一点(不填充中间数据结构，而只能通过操纵流)？ 看起来不太干净! 。谢谢。

Answer 1

您的问题似乎源于您使用flatMap的事实，因此，如果数据库中没有给定id的数据，并且您得到一个空的Observable，则flatMap不会为此类id生成任何输出。因此，看起来您需要的是defaultIfEmpty，它已转换为Scala的 orElse 。您可以使用orElse返回flatMap内部的一些默认值。因此，修改您的示例:

def fetchFromDb(id: Int): Observable[String] = {
  if (id <= 3)
    Observable.just(s"Document #$id")
  else
    Observable.empty
}

def gotValue(idsToFetch: Seq[Int]): Observable[(Int, Boolean)] = {
  Observable.from(idsToFetch).flatMap((id: Int) => fetchFromDb(id).map(_ => (id, true)).orElse((id, false)))
}

println(gotValue(Seq(1, 2, 3, 4, 5, 6)).toBlocking.toList)

哪个打印

List((1,true), (2,true), (3,true), (4,false), (5,false), (6,false))

或者，您可以使用 Option返回 Some(JsonDocument)或 None，例如

def gotValueEx(idsToFetch: Seq[Int]): Observable[(Int, Option[String])] = {
  Observable.from(idsToFetch).flatMap((id: Int) => fetchFromDb(id).map(doc => (id, Option(doc))).orElse((id, None)))
}

println(gotValueEx(Seq(1, 2, 3, 4, 5, 6)).toBlocking.toList)

哪个打印

List((1,Some(Document #1)), (2,Some(Document #2)), (3,Some(Document #3)), (4,None), (5,None), (6,None))