Reactive Scala Driver for MongoDB

Asynchronous & Non-Blocking

Find documents

Note: the following snippets of code use a BSONCollection (the default collection implementation return by db.collection()).

Perform a simple query

Queries are performed in quite the same way as in the MongoDB Shell.

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

import reactivemongo.api.bson._
import reactivemongo.api.bson.collection.BSONCollection

def findOlder1(collection: BSONCollection): Future[Option[BSONDocument]] = {
  // { "age": { "$gt": 27 } }
  val query = BSONDocument("age" -> BSONDocument("$gt" -> 27))

  // MongoDB .findOne
  collection.find(query).one[BSONDocument]
}

Of course you can collect only a limited number of documents.

import scala.concurrent.ExecutionContext.Implicits.global

import reactivemongo.api.bson.BSONDocument

import reactivemongo.api.Cursor
import reactivemongo.api.bson.collection.BSONCollection

def findOlder2(collection: BSONCollection) = {
  val query = BSONDocument("age" -> BSONDocument("$gt" -> 27))

  // only fetch the name field for the result documents
  val projection = BSONDocument("name" -> 1)

  collection.find(query, projection).cursor[BSONDocument]().
    collect[List](25, // get up to 25 documents
      Cursor.FailOnError[List[BSONDocument]]())
}

When using a serialization pack other than the BSON default one, then the appropriate document type must be used to define query (e.g. JsObject for the JSON serialization).

The find method returns a query builder (e.g. a BSONQueryBuilder), which means the query is therefore not performed yet. It gives you the opportunity to add options to the query, like a sort order, projection, flags…

import scala.concurrent.ExecutionContext.Implicits.global

import reactivemongo.api.bson.BSONDocument

import reactivemongo.api.{ Cursor, QueryOpts }
import reactivemongo.api.bson.collection.BSONCollection

def findNOlder(collection: BSONCollection, limit: Int) = {
  val querybuilder =
    collection.find(BSONDocument("age" -> BSONDocument("$gt" -> 27)))

  // Sets options before executing the query
  querybuilder.options(QueryOpts().batchSize(limit)).
    cursor[BSONDocument]().collect[List](10, // get up to 10 documents
      Cursor.FailOnError[List[BSONDocument]]())
 
}

The class QueryOpts is used to prepared the query options.

When your query is ready to be sent to MongoDB, you may just call one of the following function.

The function cursor which returns a Cursor[BSONDocument].
The function one which returns a Future[Option[T]] (the first document that matches the query, if any).
The function requireOne which returns a Future[T] with the first matching document, or fails if none.

import scala.concurrent.{ ExecutionContext, Future }
import reactivemongo.api.bson.BSONDocument
import reactivemongo.api.bson.collection.BSONCollection

trait PersonService1 {
  def collection: BSONCollection

  def requirePerson(firstName: String, lastName: String)(implicit ec: ExecutionContext): Future[Person] = collection.find(BSONDocument(
    "firstName" -> firstName,
    "lastName" -> lastName
  )).requireOne[Person]
}

On a cursor, the collect function can be used. It must be given a Scala collection type, like List or Vector. It accumulates all the results in memory.

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

import reactivemongo.api.Cursor
import reactivemongo.api.bson.collection.BSONCollection

trait PersonService2 {
  def collection: BSONCollection

  def persons(age: Int)(implicit ec: ExecutionContext): Future[List[Person]] =
    collection.find(BSONDocument("age" -> age)).cursor[Person]().
    collect[List](-1, Cursor.FailOnError[List[Person]]())
}

More: Streaming

Find and sort documents

The return type of the find method is a GenericQueryBuilder, which enables to customize the query, especially to add sort information. Like in the MongoDB console, you sort by giving a document containing the field names associated with an order (1 for ascending, -1 descending). Let’s sort our previous query by last name, in the alphabetical order (the sort document is also { lastName: 1 }).

import scala.concurrent.ExecutionContext.Implicits.global
import reactivemongo.api.bson.BSONDocument

import reactivemongo.api.Cursor
import reactivemongo.api.bson.collection.BSONCollection

def findOlder3(collection: BSONCollection) = {
  val query = BSONDocument("age" -> BSONDocument("$gt" -> 27))

  collection.find(query).
    sort(BSONDocument("lastName" -> 1)). // sort by lastName
    cursor[BSONDocument]().
    collect[List](-1, Cursor.FailOnError[List[BSONDocument]]())
}

See: Query builder

Use Readers to deserialize documents automatically

As explained here, you can use the BSONDocumentReader and BSONDocumentWriter typeclasses to handle de/serialization between BSONDocument and your model classes.

import reactivemongo.api.bson._

case class Person(
  id: String,
  firstName: String,
  lastName: String,
  age: Int)

object Person {
  implicit object PersonReader extends BSONDocumentReader[Person] {
    def readDocument(doc: BSONDocument) = for {
      id <- doc.getAsTry[BSONObjectID]("_id").map(_.stringify)
      firstName <- doc.getAsTry[String]("firstName")
      lastName <- doc.getAsTry[String]("lastName")
      age <- doc.getAsTry[Int]("age")
    } yield Person(id, firstName, lastName, age)
  }
}

This system is fully supported in the Collection API, so you can get the results of your queries in the right type.

Any error raised by the read function will be caught by ReactiveMongo deserialization, and will result in an explicit Future failure.

import scala.concurrent.{ ExecutionContext, Future }

import reactivemongo.api.bson.{ BSONDocument, BSONDocumentReader }

import reactivemongo.api.Cursor
import reactivemongo.api.bson.collection.BSONCollection

def findOlder4(collection: BSONCollection)(implicit ec: ExecutionContext, reader: BSONDocumentReader[Person]): Future[List[Person]] = {
  val query = BSONDocument("age" -> BSONDocument("$gt" -> 27))

  val peopleOlderThanTwentySeven = collection.find(query).
    /*
     * Indicate that the documents should be transformed into `Person`.
     * A `BSONDocumentReader[Person]` should be in the implicit scope.
     */
    cursor[Person](). // ... collect in a `List`
    collect[List](-1, Cursor.FailOnError[List[Person]]())

  peopleOlderThanTwentySeven.map { people =>
    for (person <- people) println(s"found $person")
  }

  peopleOlderThanTwentySeven
}

ReactiveMongo can directly return instances of a custom class, by defining a custom reader.

Troubleshooting

The synchronous .db has been deprecated as it didn’t offer a sufficient guaranty that it can initially find an active channel in the connection pool (MongoConnection). A corresponding warning is raised by the compiler in such case.

method db in class MongoConnection is deprecated: Must use [[database]]

The new .database resolution must be used (see connection tutorial).

If the deprecated database resolution is still used, a runtime error such as ConnectionNotInitialized can be raised when querying documents.

On query builder, the previous cursor has been deprecated:

Use `cursor()` or `cursor(readPreference)`

As indicated by this compilation warning, the new cursor is expecting a ReadPreference as parameter.

When a Cursor has been obtained, a warning can be raised if using the deprecated collect function:

method collect in trait Cursor is deprecated: Use `collect` with an [[Cursor.ErrorHandler]].

The new collect function must be used instead.

Due to the refactoring of the BSON API, the following warnings can be raised.

Use reactivemongo-bson-api or use reactivemongo-bson-compat: import reactivemongo.api.bson.compat._

- OR -

Use reactivemongo-bson-api

In this case either update the code to the new BSON API, or use the compatibility library to ease the migration.

import reactivemongo.api.bson.compat._

Previous: Write Documents / Next: Streaming

Suggest changes