MongoDB Scala 驱动程序 - 无法检索文档

Question

我在使用新的scala驱动程序在mongoDB中检索文档时遇到问题。

这是我的代码：

def retrieveDocument(id: Int,
  collectionName: String, 
  databaseName: String,
  url: String): Option[Document] = {
    var res: Option[Document] = None
    getMongoCollectionImmutable(collectionName, databaseName, url)
      .find(Filters.equal("_id", id))
      .first().subscribe(
      (doc: Document) => res = Some(doc),
      (e: Throwable) => throw e,
      () => ())
    res
  }
def getMongoCollectionImmutable(collectionName: String,databaseName: String, url: String = DEFAULT_URL): MongoCollection[ImmuDoc] = {
  db match {
    case None =>
      getMongoDatabase(databaseName, url).getCollection(collectionName)
    case Some(db) =>
      db.client.getDatabase(databaseName).getCollection(collectionName)
  }
def getMongoDatabase(name: String, url: String = DEFAULT_URL): MongoDatabase = {
  db match {
    case None =>
      db = Some(new _Database(url))
      getMongoDatabase(name)
    case Some(db) =>
      db.client.getDatabase(name)
  }
def retrieve(id: Int): Try[User] = {
  try {
    val docOption = Database.retrieveDocument(id, USER_COLLECTION, DATABASE_NAME, DEFAULT_URL) 
    docOption match {
      case None      => Failure(new Exception(s"Unable to retrieve a user with id ${id}"))
      case Some(doc) => Try(User(doc))
    }
  } catch {
    case e: Throwable => Failure(e)
  }
}

以下是驱动程序的（重要）日志：

19:16:24.334 DEBUG cluster - Updating cluster description to  {type=STANDALONE, servers=[{address=localhost:27017, type=STANDALONE, roundTripTime=0.7 ms, state=CONNECTED}]
19:16:24.366 INFO  connection - Opened connection [connectionId{localValue:2, serverValue:90}] to localhost:27017
19:16:24.377 DEBUG query - Asynchronously sending query of namespace jobless.user on connection [connectionId{localValue:2, serverValue:90}] to server localhost:27017
19:16:24.381 DEBUG query - Query results received 1 documents with cursor null

这是我的测试输出

Run starting. Expected test count is: 1
UserTest:
User Document((_id,BsonInt32{value=1}), (firstname,BsonString{value='user1'}), (lastname,BsonString{value='last1'}), (encryptedPass,BsonString{value='pass'}), (cvListPath,BsonArray{values=[{ "name" : "path1", "path" : "name1" }, { "name" : "path2", "path" : "name2" }]}), (motivationLettersPath,BsonArray{values=[{ "name" : "path1", "path" : "name1" }, { "name" : "path2", "path" : "name2" }]}))
-  retrieve from DB Failure(java.lang.Exception: Unable to retrieve a user with id 1) *** FAILED ***
  java.lang.Exception: Unable to retrieve a user with id 1 (UserTest.scala:31)

但！通过使用wireshark，我可以以正确的方式查看数据库查询并返回文档！（我通过使用 mongo linux 命令进行查询进行了检查，它是一样的）

Frame 1262: 123 bytes on wire (984 bits), 123 bytes captured (984 bits) on interface 0
Linux cooked capture
Internet Protocol Version 4, Src: 127.0.0.1, Dst: 127.0.0.1
Transmission Control Protocol, Src Port: 42714, Dst Port: 27017, Seq: 438, Ack: 1272, Len: 55
Mongo Wire Protocol
    Message Length: 55
    Request ID: 0x00000008 (8)
    Response To: 0x00000000 (0)
    OpCode: Query (2004)
    Query Flags
    fullCollectionName: xxx.user
    Number To Skip: 0
    Number to Return: -1
    Query
        Document length: 14
        Elements
            Element: _id
                Type: Int32 (0x10)
                Value: 1

并且数据库响应也是正确的。

我做错了什么？

Answer 1

好吧，我责怪自己（以及驱动程序的神秘文档）

这个驱动程序是异步的，所以我们只需要等待它完成......因此，总结一下使用它（在此处找到），它真正在文档中使用，而不在驱动程序本身中

implicit class DocumentObservable[C](val observable: Observable[Document]) extends ImplicitObservable[Document] {
   override val converter: (Document) => String = (doc) => doc.toJson
}
implicit class GenericObservable[C](val observable: Observable[C]) extends ImplicitObservable[C] {
   override val converter: (C) => String = (doc) => doc.toString
}
trait ImplicitObservable[C] {
  val observable: Observable[C]
  val converter: (C) => String
  def results(): Seq[C] = Await.result(observable.toFuture(), Duration(10, TimeUnit.SECONDS))
  def headResult() = Await.result(observable.head(), Duration(10, TimeUnit.SECONDS))
  def printResults(initial: String = ""): Unit = {
    if (initial.length > 0) print(initial)
    results().foreach(res => println(converter(res)))
  }
  def printHeadResult(initial: String = ""): Unit = println(s"${initial}${converter(headResult())}")
}

并以这种方式更改检索：

def retrieveDocument(id: Int, collectionName: String, databaseName: String, url: String): Option[Document] = {
var res: Option[Document] = None
getMongoCollectionImmutable(collectionName, databaseName, url)
  .find(Filters.equal("_id", id))
  .limit(1).results().foreach({ x => res = Some(x) })
res

}

它解决了我的问题。