Upsert many records using ReactiveMongo and Scala

问题:

I am writing a DAO Actor for MongoDB that uses ReactiveMongo. I want to implement some very simple CRUD operations, among which the ability to upsert many records in one shot. Since I have a reactive application (built on Akka), it's important for me to have idempotent actions, so I need the operation to be an upsert, not an insert.

So far I have the following (ugly) code to do so:

case class UpsertResult[T](nUpd: Int, nIns: Int, failed: List[T])

def upsertMany[T](l: List[T], collection: BSONCollection)
  (implicit ec: ExecutionContext, w: BSONDocumentWriter[T]):
    Future[UpsertResult[T]] = {
      Future.sequence(l.map(o => collection.save(o).map(r => (o, r))))
        .transform({
          results =>      
            val failed: List[T] = results.filter(!_._2.ok).unzip._1
            val nUpd = results.count(_._2.updatedExisting)
            UpsertResult(nUpd, results.size - nUpd - failed.size, failed)
        }, t => t)
      }    

Is there an out-of-the-box way of upserting many records at once using the reactivemongo API alone?

I am a MongoDB beginner so this might sound trivial to many. Any help is appreciated!

回答1:

Mongo has no support for upserting multiple documents in one query. The update operation for example can always only insert up to one new element. So this is not a flaw in the reactivemongo driver, there simply is no DB command to achieve the result you expect. Iterating over the documents you want to upsert is the right way to do it.

The manual on mongodb about upsert contains further informations:

http://docs.mongodb.org/manual/core/update/#update-operations-with-the-upsert-flag

回答2:

According to the docs, BSSONCollection.save inserts the document, or updates it if it already exists in the collection: see here. Now, I'm not sure exactly how it makes the decision about whether the document already exists or not: presumably it's based on what MongoDB tells it... so the primary key/id or a unique index.

In short: I think you're doing it the right way (including your result counts from LastError).