DATAMONGO-1646 - Support reactive aggregation execution.

We now support reactive aggregation execution via ReactiveMongoOperations.aggregate(…).

Original Pull Request: #481

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java

@@ -24,6 +24,9 @@ import org.bson.Document;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
 import org.springframework.data.geo.GeoResult;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
+import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
 import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.index.ReactiveIndexOperations;
@@ -370,6 +373,73 @@ public interface ReactiveMongoOperations {
 	 */
 	<T> Mono<T> findById(Object id, Class<T> entityClass, String collectionName);
 
+	/**
+	 * Execute an aggregation operation.
+	 * <p>
+	 * The raw results will be mapped to the given entity class. The name of the inputCollection is derived from the
+	 * inputType of the aggregation.
+	 * <p>
+	 * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling
+	 * explanation mode will throw an {@link IllegalArgumentException}.
+	 *
+	 * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be
+	 *          {@literal null}.
+	 * @param collectionName The name of the input collection to use for the aggreation.
+	 * @param outputType The parametrized type of the returned list, must not be {@literal null}.
+	 * @return The results of the aggregation operation.
+	 */
+	<O> Flux<O> aggregate(TypedAggregation<?> aggregation, String collectionName, Class<O> outputType);
+
+	/**
+	 * Execute an aggregation operation.
+	 * <p/>
+	 * The raw results will be mapped to the given entity class and are returned as stream. The name of the
+	 * inputCollection is derived from the inputType of the aggregation.
+	 * <p/>
+	 * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling
+	 * explanation mode will throw an {@link IllegalArgumentException}.
+	 *
+	 * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be
+	 *          {@literal null}.
+	 * @param outputType The parametrized type of the returned list, must not be {@literal null}.
+	 * @return The results of the aggregation operation.
+	 */
+	<O> Flux<O> aggregate(TypedAggregation<?> aggregation, Class<O> outputType);
+
+	/**
+	 * Execute an aggregation operation.
+	 * <p/>
+	 * The raw results will be mapped to the given entity class.
+	 * <p/>
+	 * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling
+	 * explanation mode will throw an {@link IllegalArgumentException}.
+	 *
+	 * @param aggregation The {@link Aggregation} specification holding the aggregation operations, must not be
+	 *          {@literal null}.
+	 * @param inputType the inputType where the aggregation operation will read from, must not be {@literal null} or
+	 *          empty.
+	 * @param outputType The parametrized type of the returned list, must not be {@literal null}.
+	 * @return The results of the aggregation operation.
+	 */
+	<O> Flux<O> aggregate(Aggregation aggregation, Class<?> inputType, Class<O> outputType);
+
+	/**
+	 * Execute an aggregation operation.
+	 * <p/>
+	 * The raw results will be mapped to the given entity class.
+	 * <p/>
+	 * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling
+	 * explanation mode will throw an {@link IllegalArgumentException}.
+	 *
+	 * @param aggregation The {@link Aggregation} specification holding the aggregation operations, must not be
+	 *          {@literal null}.
+	 * @param collectionName the collection where the aggregation operation will read from, must not be {@literal null} or
+	 *          empty.
+	 * @param outputType The parametrized type of the returned list, must not be {@literal null}.
+	 * @return The results of the aggregation operation.
+	 */
+	<O> Flux<O> aggregate(Aggregation aggregation, String collectionName, Class<O> outputType);
+
 	/**
 	 * Returns {@link Flux} of {@link GeoResult} for all entities matching the given {@link NearQuery}. Will consider
 	 * entity mapping information to determine the collection the query is ran against. Note, that MongoDB limits the

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -65,6 +65,11 @@ import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.data.mongodb.MongoDbFactory;
 import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
+import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
+import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
+import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
 import org.springframework.data.mongodb.core.convert.DbRefProxyHandler;
 import org.springframework.data.mongodb.core.convert.DbRefResolver;
 import org.springframework.data.mongodb.core.convert.DbRefResolverCallback;
@@ -119,6 +124,7 @@ import com.mongodb.client.model.ReturnDocument;
 import com.mongodb.client.model.UpdateOptions;
 import com.mongodb.client.result.DeleteResult;
 import com.mongodb.client.result.UpdateResult;
+import com.mongodb.reactivestreams.client.AggregatePublisher;
 import com.mongodb.reactivestreams.client.FindPublisher;
 import com.mongodb.reactivestreams.client.MongoClient;
 import com.mongodb.reactivestreams.client.MongoCollection;
@@ -628,6 +634,84 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		return doFindOne(collectionName, new Document(idKey, id), null, entityClass, null);
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.TypedAggregation, java.lang.String, java.lang.Class)
+	 */
+	@Override
+	public <O> Flux<O> aggregate(TypedAggregation<?> aggregation, String inputCollectionName, Class<O> outputType) {
+
+		Assert.notNull(aggregation, "Aggregation pipeline must not be null!");
+
+		AggregationOperationContext context = new TypeBasedAggregationOperationContext(aggregation.getInputType(),
+				mappingContext, queryMapper);
+		return aggregate(aggregation, inputCollectionName, outputType, context);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.TypedAggregation, java.lang.Class)
+	 */
+	@Override
+	public <O> Flux<O> aggregate(TypedAggregation<?> aggregation, Class<O> outputType) {
+		return aggregate(aggregation, determineCollectionName(aggregation.getInputType()), outputType);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.Aggregation, java.lang.Class, java.lang.Class)
+	 */
+	@Override
+	public <O> Flux<O> aggregate(Aggregation aggregation, Class<?> inputType, Class<O> outputType) {
+
+		return aggregate(aggregation, determineCollectionName(inputType), outputType,
+				new TypeBasedAggregationOperationContext(inputType, mappingContext, queryMapper));
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.Aggregation, java.lang.String, java.lang.Class)
+	 */
+	@Override
+	public <O> Flux<O> aggregate(Aggregation aggregation, String collectionName, Class<O> outputType) {
+		return aggregate(aggregation, collectionName, outputType, null);
+	}
+
+	protected <O> Flux<O> aggregate(Aggregation aggregation, String collectionName, Class<O> outputType,
+			AggregationOperationContext context) {
+
+		Assert.hasText(collectionName, "Collection name must not be null or empty!");
+		Assert.notNull(aggregation, "Aggregation pipeline must not be null!");
+		Assert.notNull(outputType, "Output type must not be null!");
+
+		AggregationOperationContext rootContext = context == null ? Aggregation.DEFAULT_CONTEXT : context;
+
+		Document command = aggregation.toDocument(collectionName, rootContext);
+
+		Boolean explain = command.get("explain", Boolean.class);
+		if (explain != null && explain) {
+			throw new IllegalArgumentException("Can't use explain option with streaming!");
+		}
+
+		if (LOGGER.isDebugEnabled()) {
+			LOGGER.debug("Streaming aggregation: {}", serializeToJsonSafely(command));
+		}
+
+		ReadDocumentCallback<O> readCallback = new ReadDocumentCallback<>(mongoConverter, outputType, collectionName);
+
+		return execute(collectionName, collection -> {
+
+			List<Document> pipeline = (List<Document>) command.get("pipeline");
+
+			AggregationOptions options = AggregationOptions.fromDocument(command);
+
+			AggregatePublisher<Document> cursor = collection.aggregate(pipeline).allowDiskUse(options.isAllowDiskUse())
+					.useCursor(true);
+
+			if (options.getCollation().isPresent()) {
+				cursor = cursor.collation(options.getCollation().map(Collation::toMongoCollation).get());
+			}
+
+			return Flux.from(cursor).map(readCallback::doWith);
+		});
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#geoNear(org.springframework.data.mongodb.core.query.NearQuery, java.lang.Class)

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/City.java

@@ -1,5 +1,11 @@
 package org.springframework.data.mongodb.core.aggregation;
 
+import lombok.AllArgsConstructor;
+import lombok.NoArgsConstructor;
+
+@lombok.Data
+@AllArgsConstructor
+@NoArgsConstructor
 class City {
 
 	String name;

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationTests.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.core.aggregation;
+
+import static org.assertj.core.api.AssertionsForInterfaceTypes.*;
+import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
+import static org.springframework.data.mongodb.core.query.Criteria.*;
+
+import reactor.test.StepVerifier;
+
+import java.util.Arrays;
+
+import org.bson.Document;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * Integration test for aggregation via {@link org.springframework.data.mongodb.core.ReactiveMongoTemplate}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration("classpath:reactive-infrastructure.xml")
+public class ReactiveAggregationTests {
+
+	private static final String INPUT_COLLECTION = "aggregation_test_collection";
+	private static final String OUTPUT_COLLECTION = "aggregation_test_out";
+
+	@Autowired ReactiveMongoTemplate reactiveMongoTemplate;
+
+	@Before
+	public void setUp() {
+		cleanDb();
+	}
+
+	@After
+	public void cleanUp() {
+		cleanDb();
+	}
+
+	private void cleanDb() {
+		StepVerifier
+				.create(reactiveMongoTemplate.dropCollection(INPUT_COLLECTION) //
+						.then(reactiveMongoTemplate.dropCollection(OUTPUT_COLLECTION)) //
+						.then(reactiveMongoTemplate.dropCollection(Product.class)) //
+						.then(reactiveMongoTemplate.dropCollection(City.class))) //
+				.verifyComplete();
+	}
+
+	@Test(expected = IllegalArgumentException.class) // DATAMONGO-1646
+	public void shouldHandleMissingInputCollection() {
+		reactiveMongoTemplate.aggregate(newAggregation(), (String) null, TagCount.class);
+	}
+
+	@Test(expected = IllegalArgumentException.class) // DATAMONGO-1646
+	public void shouldHandleMissingAggregationPipeline() {
+		reactiveMongoTemplate.aggregate(null, INPUT_COLLECTION, TagCount.class);
+	}
+
+	@Test(expected = IllegalArgumentException.class) // DATAMONGO-1646
+	public void shouldHandleMissingEntityClass() {
+		reactiveMongoTemplate.aggregate(newAggregation(), INPUT_COLLECTION, null);
+	}
+
+	@Test // DATAMONGO-1646
+	public void expressionsInProjectionExampleShowcase() {
+
+		Product product = new Product("P1", "A", 1.99, 3, 0.05, 0.19);
+		StepVerifier.create(reactiveMongoTemplate.insert(product)).expectNextCount(1).verifyComplete();
+
+		double shippingCosts = 1.2;
+
+		TypedAggregation<Product> agg = newAggregation(Product.class, //
+				project("name", "netPrice") //
+						.andExpression("netPrice * 10", shippingCosts).as("salesPrice") //
+		);
+
+		StepVerifier.create(reactiveMongoTemplate.aggregate(agg, Document.class)).consumeNextWith(actual -> {
+
+			assertThat(actual).containsEntry("_id", product.id);
+			assertThat(actual).containsEntry("name", product.name);
+			assertThat(actual).containsEntry("salesPrice", product.netPrice * 10);
+		}).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1646
+	public void shouldProjectMultipleDocuments() {
+
+		City dresden = new City("Dresden", 100);
+		City linz = new City("Linz", 101);
+		City braunschweig = new City("Braunschweig", 102);
+		City weinheim = new City("Weinheim", 103);
+
+		StepVerifier.create(reactiveMongoTemplate.insertAll(Arrays.asList(dresden, linz, braunschweig, weinheim)))
+				.expectNextCount(4).verifyComplete();
+
+		Aggregation agg = newAggregation( //
+				match(where("population").lt(103)));
+
+		StepVerifier.create(reactiveMongoTemplate.aggregate(agg, "city", City.class).collectList())
+				.consumeNextWith(actual -> {
+					assertThat(actual).hasSize(3).contains(dresden, linz, braunschweig);
+				}).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1646
+	public void shouldAggregateToOutCollection() {
+
+		City dresden = new City("Dresden", 100);
+		City linz = new City("Linz", 101);
+		City braunschweig = new City("Braunschweig", 102);
+		City weinheim = new City("Weinheim", 103);
+
+		StepVerifier.create(reactiveMongoTemplate.insertAll(Arrays.asList(dresden, linz, braunschweig, weinheim)))
+				.expectNextCount(4).verifyComplete();
+
+		Aggregation agg = newAggregation( //
+				out(OUTPUT_COLLECTION));
+
+		StepVerifier.create(reactiveMongoTemplate.aggregate(agg, "city", City.class)).expectNextCount(4).verifyComplete();
+		StepVerifier.create(reactiveMongoTemplate.find(new Query(), City.class, OUTPUT_COLLECTION)).expectNextCount(4)
+				.verifyComplete();
+	}
+}