0

我正在使用带有 Scala、Akka 和 ReactiveMongo 的 Play Framework。我想将 MongoDB 中的集合用作循环队列。多个参与者可以将文档插入其中;一个参与者在这些文档可用时立即检索它们(一种发布-订阅系统)。我正在使用上限集合和可尾光标。每次我检索一些文档时,我都必须运行命令 EmptyCapped 来刷新上限集合(不可能从中删除元素),否则我总是检索相同的文档。有替代解决方案吗?例如,有没有办法在不删除元素的情况下滑动光标?或者在我的情况下最好不要使用上限集合?

object MexDB {

def db: reactivemongo.api.DB = ReactiveMongoPlugin.db
val size: Int = 10000

// creating capped collection
val collection: JSONCollection = {

    val c = db.collection[JSONCollection]("messages")

    val isCapped = coll.convertToCapped(size, None)

    Await.ready(isCapped, Duration.Inf)

    c
}

def insert(mex: Mex) = {

    val inserted = collection.insert(mex)

    inserted onComplete {
      case Failure(e) =>
        Logger.info("Error while inserting task: " + e.getMessage())
        throw e

      case Success(i) =>
        Logger.info("Successfully inserted task")
    }

}


def find(): Enumerator[Mex] = {

  val cursor: Cursor[Mex] = collection
    .find(Json.obj())
    .options(QueryOpts().tailable.awaitData)
    .cursor[Mex]

    // meaning of maxDocs ???
    val maxDocs = 1
    cursor.enumerate(maxDocs)
}


def removeAll() = {
    db.command(new EmptyCapped("messages"))
}

}

/*** part of receiver actor code ***/

// inside preStart
val it = Iteratee.fold[Mex, List[Mex]](Nil) {
    (partialList, mex) => partialList ::: List(mex)
}

// Inside "receive" method
case Data =>

  val e: Enumerator[Mex] = MexDB.find()

  val future = e.run(it)

  future onComplete {
    case Success(list) =>
      list foreach { mex =>
        Logger.info("Mex: " + mex.id)
      }
      MexDB.removeAll()
      self ! Data

    case Failure(e) => Logger.info("Error:  "+ e.getMessage())
  }
4

1 回答 1

0

在每个找到的文档作为maxDocs = 1. 要无限期地保持打开状态,您应该忽略此限制。

仅当您明确关闭 RMawaitData.onComplete才会调用 , 。

您需要使用光标中的一些流功能,例如.enumerate并处理每个新步骤/结果。见https://github.com/sgodbillon/reactivemongo-tailablecursor-demo/

于 2015-05-19T07:29:00.637 回答