ReactiveMongo aggregation pipeline example #2

gistfile1.scala

collection.db.command(Aggregate(collection.name, Seq(
      Match(toBSON(jsonQuery)).get.asInstanceOf[BSONDocument]),
      Project(convertProjectionJsonForAggregate(projectionJson):_*),
      Sort(convertSortJsonForAggregate(versionDescendingSort))
    ))) map { stream =>
        stream.toList map { doc =>
            toJSON(doc).asInstanceOf[JsObject]
        }
    }

  def convertProjectionJsonForAggregate(projectionJson: JsObject): Seq[(String, Int)] = {
    projectionJson.fields filter {
      (kv: (String, JsValue)) => kv._2.isInstanceOf[JsNumber]
    } map {
      (kv: (String, JsValue)) => (kv._1, kv._2.as[Int])
    }
  }

  def convertSortJsonForAggregate(sortJson: JsObject): Seq[SortOrder] = {
    sortJson.fields filter {
      (kv: (String, JsValue)) => kv._2.as[Int] == 1 || kv._2.as[Int] == -1
    } map {
      (kv: (String, JsValue)) => if (kv._2.as[Int] == 1) Ascending(kv._1) else Descending(kv._1)
    }
  }