带有 Play Framework 2.5 的 ReactiveMongo 数据库转储

Question

我正在尝试将我的 mongo 数据库转储到一个 json 对象中，但是因为我对数据库的查询是异步的，所以我遇到了问题。

我的数据库中的每个集合都包含用户数据，每个集合名称都是一个用户名。

因此，当我想获取所有用户数据时，我会恢复所有集合名称，然后遍历它们以逐个恢复每个集合。

def databaseDump(prom : Promise[JsObject]) = {
    for{
      dbUsers <- getUsers
    } yield dbUsers

var rebuiltJson = Json.obj()
var array = JsArray()
res.map{ users =>
  users.map{ userNames =>
    if(userNames.size == 0){
      prom failure new Throwable("Empty database")
    }
    var counter = 0
    userNames.foreach { username =>
      getUserTables(username).map { tables =>
         /* Add data to array*/
           ...
        counter += 1
        if(counter == userNames.size){
          /*Add data to new json*/
             ...
          prom success rebuiltJson
        }

      } 
    }
  }
}

这有点工作，但有时即使所有数据尚未恢复，promise 也会成功触发。这是因为我的计数器变量不是一个可靠的解决方案。

有没有办法遍历所有用户，查询数据库并等待所有数据恢复，然后才能成功触发承诺？我试图用于理解，但没有找到方法。有没有办法将整个 mongo DB 转储到一个 Json 中：{ username : data, username : data ..}？

Answer 1

用户/表术语让我感到困惑，因此我编写了一个新函数，将数据库转储到单个 JsObject。

// helper function to find all documents inside a collection c
// and return them as a single JsArray
def getDocs(c: JSONCollection)(implicit ec: ExecutionContext) = c.find(Json.obj()).cursor[JsObject]().jsArray()

def dumpToJsObject(db: DefaultDB)(implicit ec: ExecutionContext): Future[JsObject] = {
  // get a list of all collections in the db
  val collectionNames = db.collectionNames
  val collections = collectionNames.map(_.map(db.collection[JSONCollection](_)))

  // each entry is a tuple collectionName -> content (as JsArray)
  val namesToDocs = collections.flatMap {
    colls => Future.sequence(colls.map(c => getDocs(c).map(c.name -> _)))
  }

  // convert to a single JsObject
  namesToDocs.map(JsObject(_))
}

我还没有测试它（我稍后会这样做），但这个函数至少应该给你一个大致的想法。您将获得数据库内所有集合的列表。对于每个集合，您执行查询以获取该集合中的所有文档。将文档列表转换为JsArray，最后将所有集合组合成单个JsObject，集合名称为键。

Answer 2

如果目标是将数据写入输出流（本地/文件或网络），则会产生副作用。

import scala.concurrent.{ ExecutionContext, Future }
import reactivemongo.bson.BSONDocument
import reactivemongo.api.{ Cursor, MongoDriver, MongoConnection }

val mongoUri = "mongodb://localhost:27017/my_db"

val driver = new MongoDriver
val maxDocs = Int.MaxValue // max per collection

// Requires to have an ExecutionContext in the scope
// (e.g. `import scala.concurrent.ExecutionContext.Implicits.global`)
def dump()(implicit ec: ExecutionContext): Future[Unit] = for {
  uri <- Future.fromTry(MongoConnection.parseURI(mongoUri))
  con = driver.connection(uri)
  dn <- Future(uri.db.get)
  db <- con.database(dn)
  cn <- db.collectionNames
  _  <- Future.sequence(cn.map { collName =>
    println(s"Collection: $collName")

    db.collection(collName).find(BSONDocument.empty). // findAll
      cursor[BSONDocument]().foldWhile({}, maxDocs) { (_, doc) =>
        // Replace println by appropriate side-effect
        Cursor.Cont(println(s"- ${BSONDocument pretty doc}"))
      }
  })
} yield ()

如果使用 JSON 序列化包，只需将 BSONDocument 替换为 JsObject（例如 BSONDocument.empty ~> Json.obj()）。

如果从 Scala REPL 测试，粘贴之前的代码后，可以如下执行。

dump().onComplete {
  case result =>
    println(s"Dump result: $result")
    //driver.close()
}