[DATAMONGO-2655] Add missing aggregate extension

spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveMongoOperationsExtensions.kt

@@ -20,6 +20,8 @@ import com.mongodb.client.result.UpdateResult
 import com.mongodb.reactivestreams.client.MongoCollection
 import org.bson.Document
 import org.springframework.data.geo.GeoResult
+import org.springframework.data.mongodb.core.aggregation.Aggregation
+import org.springframework.data.mongodb.core.aggregation.TypedAggregation
 import org.springframework.data.mongodb.core.index.ReactiveIndexOperations
 import org.springframework.data.mongodb.core.query.NearQuery
 import org.springframework.data.mongodb.core.query.Query
@@ -210,6 +212,43 @@ inline fun <reified T : Any, reified E : Any> ReactiveMongoOperations.findDistin
 		if (collectionName != null) findDistinct(query, field, collectionName, E::class.java, T::class.java)
 		else findDistinct(query, field, E::class.java, T::class.java)
 
+
+/**
+ * Extension for [ReactiveMongoOperations.aggregate] leveraging reified type parameters.
+ *
+ * @author Wonwoo Lee
+ * @since 3.1
+ */
+inline fun <reified T : Any> ReactiveMongoOperations.aggregate(aggregation: TypedAggregation<*>, collectionName: String): Flux<T> =
+		this.aggregate(aggregation, collectionName, T::class.java)
+
+/**
+ * Extension for [ReactiveMongoOperations.aggregate] leveraging reified type parameters.
+ *
+ * @author Wonwoo Lee
+ * @since 3.1
+ */
+inline fun <reified T : Any> ReactiveMongoOperations.aggregate(aggregation: TypedAggregation<*>): Flux<T> =
+		this.aggregate(aggregation, T::class.java)
+
+/**
+ * Extension for [ReactiveMongoOperations.aggregate] leveraging reified type parameters.
+ *
+ * @author Wonwoo Lee
+ * @since 3.1
+ */
+inline fun <reified T : Any> ReactiveMongoOperations.aggregate(aggregation: Aggregation, entityClass: KClass<*>): Flux<T> =
+		this.aggregate(aggregation, entityClass.java, T::class.java)
+
+/**
+ * Extension for [ReactiveMongoOperations.aggregate] leveraging reified type parameters.
+ *
+ * @author Wonwoo Lee
+ * @since 3.1
+ */
+inline fun <reified T : Any> ReactiveMongoOperations.aggregate(aggregation: Aggregation, collectionName: String): Flux<T> =
+		this.aggregate(aggregation, collectionName, T::class.java)
+
 /**
  * Extension for [ReactiveMongoOperations.geoNear] leveraging reified type parameters.
  *

spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveMongoOperationsExtensionsTests.kt

@@ -19,6 +19,8 @@ import example.first.First
 import io.mockk.mockk
 import io.mockk.verify
 import org.junit.Test
+import org.springframework.data.mongodb.core.aggregation.Aggregation
+import org.springframework.data.mongodb.core.aggregation.TypedAggregation
 import org.springframework.data.mongodb.core.query.NearQuery
 import org.springframework.data.mongodb.core.query.Query
 import org.springframework.data.mongodb.core.query.Update
@@ -598,7 +600,6 @@ class ReactiveMongoOperationsExtensionsTests {
 		verify { operations.findDistinct(query, "field", "collection", First::class.java, String::class.java) }
 	}
 
-
 	@Test // DATAMONGO-1761
 	@Suppress("DEPRECATION")
 	fun `findDistinct(Query, String,  KClass) should call java counterpart`() {
@@ -608,4 +609,40 @@ class ReactiveMongoOperationsExtensionsTests {
 		operations.findDistinct<String>(query, "field", First::class)
 		verify { operations.findDistinct(query, "field", First::class.java, String::class.java) }
 	}
+
+	@Test
+	fun `aggregate(TypedAggregation, String, KClass) should call java counterpart`() {
+
+		val aggregation = mockk<TypedAggregation<String>>()
+
+		operations.aggregate<First>(aggregation, "foo")
+		verify { operations.aggregate(aggregation, "foo", First::class.java) }
+	}
+
+	@Test
+	fun `aggregate(TypedAggregation, KClass) should call java counterpart`() {
+
+		val aggregation = mockk<TypedAggregation<String>>()
+
+		operations.aggregate<First>(aggregation)
+		verify { operations.aggregate(aggregation, First::class.java) }
+	}
+
+	@Test
+	fun `aggregate(Aggregation, KClass) should call java counterpart`() {
+
+		val aggregation = mockk<Aggregation>()
+
+		operations.aggregate<First>(aggregation, String::class)
+		verify { operations.aggregate(aggregation, String::class.java, First::class.java) }
+	}
+
+	@Test
+	fun `aggregate(Aggregation, String) should call java counterpart`() {
+
+		val aggregation = mockk<Aggregation>()
+
+		operations.aggregate<First>(aggregation, "foo")
+		verify { operations.aggregate(aggregation, "foo", First::class.java) }
+	}
 }