Reactive Scala Driver for MongoDB

Asynchronous & Non-Blocking

Custom Aggregation Stage

You can also implement custom aggregate stage, using the PipelineOperator factory.

import scala.concurrent.ExecutionContext

import reactivemongo.bson._
import reactivemongo.api.collections.bson.BSONCollection

def customAgg(coll: BSONCollection)(implicit ec: ExecutionContext) =
  coll.aggregateWith1[BSONDocument]() { framework =>
    import framework.PipelineOperator

    val customStage = // { $sample: { size: 2 } }
      PipelineOperator(BSONDocument("$sample" -> BSONDocument("size" -> 2)))

    customStage -> List.empty
  }

Previous: Aggregation Framework