feat: Pagination for MongoDb Search and Aggregation

Author: QuadStingray

src/main/scala/dev/mongocamp/driver/mongodb/exception/MongoCampPaginationException.scala

@@ -0,0 +1,4 @@
+package dev.mongocamp.driver.mongodb.exception
+
+case class MongoCampPaginationException(message: String) extends Exception(message)
+

src/main/scala/dev/mongocamp/driver/mongodb/operation/Base.scala

@@ -14,8 +14,9 @@ abstract class Base[A]()(implicit ct: ClassTag[A]) extends LazyLogging {
 
   protected def coll: MongoCollection[A]
 
-  def count(filter: Bson = Document(), options: CountOptions = CountOptions()): Observable[Long] =
+  def count(filter: Bson = Document(), options: CountOptions = CountOptions()): Observable[Long] = {
     coll.countDocuments(filter, options)
+  }
 
   def drop(): Observable[Void] = coll.drop()
 

src/main/scala/dev/mongocamp/driver/mongodb/pagination/MongoPaginatedAggregation.scala

@@ -0,0 +1,60 @@
+package dev.mongocamp.driver.mongodb.pagination
+
+import com.mongodb.client.model.Facet
+import dev.mongocamp.driver.mongodb.exception.MongoCampPaginationException
+import dev.mongocamp.driver.mongodb.{MongoDAO, _}
+import org.mongodb.scala.bson.Document
+import org.mongodb.scala.bson.conversions.Bson
+import org.mongodb.scala.model.Aggregates
+
+import scala.jdk.CollectionConverters._
+
+case class MongoPaginatedAggregation[A <: Any](
+    dao: MongoDAO[A],
+    aggregationPipeline: List[Bson] = List(),
+    allowDiskUse: Boolean = false,
+) {
+
+  private val AggregationKeyMetaData      = "metadata"
+  private val AggregationKeyData          = "data"
+  private val AggregationKeyMetaDataTotal = "total"
+
+  def paginate(page: Long, rows: Long): PaginationResult[org.bson.BsonDocument] = {
+    if (rows <= 0) {
+      throw MongoCampPaginationException("rows per page must be greater then 0.")
+    }
+    if (page <= 0) {
+      throw MongoCampPaginationException("page must be greater then 0.")
+    }
+
+    val skip = (page - 1) * rows
+
+    val listOfMetaData: List[Bson] = List(Map("$count" -> AggregationKeyMetaDataTotal))
+    val listOfPaging: List[Bson]   = List(Map("$skip" -> skip), Map("$limit" -> rows))
+
+    val pipeline =
+      aggregationPipeline ++ List(
+        Aggregates.facet(new Facet(AggregationKeyMetaData, listOfMetaData.asJava), new Facet(AggregationKeyData, listOfPaging.asJava))
+      )
+
+    val dbResponse = dao.findAggregated(pipeline, allowDiskUse).result().asInstanceOf[Document]
+
+    val count: Long = dbResponse.get(AggregationKeyMetaData).get.asArray().get(0).asDocument().get(AggregationKeyMetaDataTotal).asNumber().longValue()
+    val allPages    = Math.ceil(count.toDouble / rows).toInt
+    val list        = dbResponse.get("data").get.asArray().asScala.map(_.asDocument())
+    PaginationResult(list.toList, PaginationInfo(count, rows, page, allPages))
+  }
+
+  def countResult: Long = {
+    val listOfMetaData: List[Bson] = List(Map("$count" -> AggregationKeyMetaDataTotal))
+    val listOfPaging: List[Bson]   = List(Map("$skip" -> 0), Map("$limit" -> 1))
+
+    val pipeline = aggregationPipeline ++ List(
+      Aggregates.facet(new Facet(AggregationKeyMetaData, listOfMetaData.asJava), new Facet(AggregationKeyData, listOfPaging.asJava))
+    )
+    val dbResponse  = dao.findAggregated(pipeline, allowDiskUse).result().asInstanceOf[Document]
+    val count: Long = dbResponse.get(AggregationKeyMetaData).get.asArray().get(0).asDocument().get(AggregationKeyMetaDataTotal).asNumber().longValue()
+    count
+  }
+
+}

src/main/scala/dev/mongocamp/driver/mongodb/pagination/MongoPaginatedFilter.scala

@@ -0,0 +1,25 @@
+package dev.mongocamp.driver.mongodb.pagination
+
+import dev.mongocamp.driver.mongodb.exception.MongoCampPaginationException
+import dev.mongocamp.driver.mongodb.{MongoDAO, _}
+import org.mongodb.scala.bson.conversions.Bson
+
+case class MongoPaginatedFilter[A <: Any](dao: MongoDAO[A], filter: Bson = Map(), sort: Bson = Map(), projection: Bson = Map()) {
+
+  def paginate(page: Long, rows: Long): PaginationResult[A] = {
+    val count = countResult
+    if (rows <= 0) {
+      throw MongoCampPaginationException("rows per page must be greater then 0.")
+    }
+    if (page <= 0) {
+      throw MongoCampPaginationException("page must be greater then 0.")
+    }
+    val allPages     = Math.ceil(count.toDouble / rows).toInt
+    val skip         = (page - 1) * rows
+    val responseList = dao.find(filter, sort, projection, rows.toInt).skip(skip.toInt).resultList()
+    PaginationResult(responseList, PaginationInfo(count, rows, page, allPages))
+  }
+
+  def countResult: Long = dao.count(filter).result()
+
+}

src/main/scala/dev/mongocamp/driver/mongodb/pagination/PaginationInfo.scala

@@ -0,0 +1,3 @@
+package dev.mongocamp.driver.mongodb.pagination
+
+case class PaginationInfo(allCount: Long, perPage: Long, page: Long, pagesCount: Long)

src/main/scala/dev/mongocamp/driver/mongodb/pagination/PaginationResult.scala

@@ -0,0 +1,3 @@
+package dev.mongocamp.driver.mongodb.pagination
+
+case class PaginationResult[A <: Any](databaseObjects: List[A], paginationInfo: PaginationInfo)

src/test/scala/dev/mongocamp/driver/mongodb/pagination/PaginationAggregationSpec.scala

@@ -0,0 +1,68 @@
+package dev.mongocamp.driver.mongodb.pagination
+
+// #agg_imports
+import dev.mongocamp.driver.mongodb.Aggregate._
+import dev.mongocamp.driver.mongodb.bson.BsonConverter
+import dev.mongocamp.driver.mongodb.dao.PersonSpecification
+// #agg_imports
+
+import dev.mongocamp.driver.mongodb.test.TestDatabase._
+import org.mongodb.scala.bson.conversions.Bson
+import org.mongodb.scala.model.Aggregates.{filter, group, sort}
+import org.mongodb.scala.model.Filters.{and, equal}
+
+class PaginationAggregationSpec extends PersonSpecification {
+
+  // #agg_stages
+  val filterStage: Bson = filter(and(equal("gender", "female"), notNullFilter("balance")))
+
+  val groupStage: Bson = group(Map("age" -> "$age"), sumField("balance"), firstField("age"))
+
+  val sortStage: Bson = sort(sortByKey("age"))
+  // #agg_stages
+
+  "Search" should {
+
+    "support aggregation filter" in {
+
+      val pipeline = List(filterStage, sortStage)
+
+      val pagination = MongoPaginatedAggregation(PersonDAO.Raw, pipeline, allowDiskUse = true)
+
+      val page = pagination.paginate(1, 10)
+
+      (page.paginationInfo.allCount must be).equalTo(98)
+
+      (page.paginationInfo.pagesCount must be).equalTo(10)
+
+      (page.databaseObjects.size must be).equalTo(10)
+    }
+
+    "support aggregation filter and group" in {
+      // #agg_execute
+      val pipeline = List(filterStage, groupStage, sortStage)
+
+      val pagination = MongoPaginatedAggregation(PersonDAO.Raw, pipeline, allowDiskUse = true)
+
+      val page = pagination.paginate(1, 10)
+
+      // #agg_execute
+      (page.paginationInfo.allCount must be).equalTo(21)
+
+      (page.paginationInfo.pagesCount must be).equalTo(3)
+
+      (page.databaseObjects.size must be).equalTo(10)
+
+      // #agg_convert
+      val list: List[Map[String, Any]] = page.databaseObjects.map(d => BsonConverter.asMap(d))
+      // #agg_convert
+      list.foreach(m => println(m("age").toString + " -> " + m("balance")))
+
+      (list.head("age") must be).equalTo(20)
+      (list.head("balance") must be).equalTo(8333.0)
+
+    }
+
+  }
+
+}

src/test/scala/dev/mongocamp/driver/mongodb/pagination/PaginationFilterSpec.scala

@@ -0,0 +1,57 @@
+package dev.mongocamp.driver.mongodb.pagination
+
+import dev.mongocamp.driver.MongoImplicits
+import dev.mongocamp.driver.mongodb.Sort._
+import dev.mongocamp.driver.mongodb._
+import dev.mongocamp.driver.mongodb.dao.PersonSpecification
+import dev.mongocamp.driver.mongodb.test.TestDatabase._
+
+class PaginationFilterSpec extends PersonSpecification with MongoImplicits {
+
+  "Search Operations" should {
+
+    "support findAll" in {
+
+      val pagination = MongoPaginatedFilter(PersonDAO)
+
+      val page = pagination.paginate(1, 10)
+
+      (page.paginationInfo.allCount must be).equalTo(PersonDAO.count().result().toInt)
+
+      page.databaseObjects.size must beEqualTo(10)
+
+      page.databaseObjects.head.name must not beEmpty
+
+      page.databaseObjects.head._id.toString must not beEmpty
+
+    }
+
+    "support with Filter" in {
+      val paginationFemale = MongoPaginatedFilter(PersonDAO, Map("gender" -> "female"), sortByKey("name"))
+
+      val pageFemale = paginationFemale.paginate(1, 10)
+
+      pageFemale.paginationInfo.pagesCount mustEqual 10
+      pageFemale.paginationInfo.allCount mustEqual 98
+      pageFemale.paginationInfo.page mustEqual 1
+      pageFemale.paginationInfo.perPage mustEqual 10
+
+      pageFemale.databaseObjects.size mustEqual 10
+      pageFemale.databaseObjects.head.name mustEqual "Adele Melton"
+
+      val paginationMales = MongoPaginatedFilter(PersonDAO, Map("gender" -> "male"))
+      val pageMale        = paginationMales.paginate(1, 10)
+
+      pageMale.paginationInfo.pagesCount mustEqual 11
+      pageMale.paginationInfo.allCount mustEqual 102
+      pageMale.paginationInfo.page mustEqual 1
+      pageMale.paginationInfo.perPage mustEqual 10
+
+      pageMale.databaseObjects.size mustEqual 10
+      pageMale.databaseObjects.head.name mustEqual "Bowen Leon"
+
+    }
+
+  }
+
+}