diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java index e7500d8ea..79b6dfd19 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java @@ -322,7 +322,7 @@ public interface MongoOperations extends FluentMongoOperations { * If your collection does not contain a homogeneous collection of types, this operation will not be an efficient way * to map objects since the test for class type is done in the client and not on the server. * - * @param entityClass the parameterized type of the returned list + * @param entityClass the parametrized type of the returned list * @return the converted collection */ List findAll(Class entityClass); @@ -336,7 +336,7 @@ public interface MongoOperations extends FluentMongoOperations { * If your collection does not contain a homogeneous collection of types, this operation will not be an efficient way * to map objects since the test for class type is done in the client and not on the server. * - * @param entityClass the parameterized type of the returned list. + * @param entityClass the parametrized type of the returned list. * @param collectionName name of the collection to retrieve the objects from * @return the converted collection */ @@ -351,7 +351,7 @@ public interface MongoOperations extends FluentMongoOperations { * @param inputCollectionName the collection where the group operation will read from * @param groupBy the conditions under which the group operation will be performed, e.g. keys, initial document, * reduce function. - * @param entityClass The parameterized type of the returned list + * @param entityClass The parametrized type of the returned list * @return The results of the group operation */ GroupByResults group(String inputCollectionName, GroupBy groupBy, Class entityClass); @@ -366,7 +366,7 @@ public interface MongoOperations extends FluentMongoOperations { * @param inputCollectionName the collection where the group operation will read from * @param groupBy the conditions under which the group operation will be performed, e.g. keys, initial document, * reduce function. - * @param entityClass The parameterized type of the returned list + * @param entityClass The parametrized type of the returned list * @return The results of the group operation */ GroupByResults group(Criteria criteria, String inputCollectionName, GroupBy groupBy, Class entityClass); @@ -378,7 +378,7 @@ public interface MongoOperations extends FluentMongoOperations { * @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 parameterized type of the returned list, 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. * @since 1.3 */ @@ -390,7 +390,7 @@ public interface MongoOperations extends FluentMongoOperations { * * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be * {@literal null}. - * @param outputType The parameterized type of the returned list, 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. * @since 1.3 */ @@ -403,7 +403,7 @@ public interface MongoOperations extends FluentMongoOperations { * {@literal null}. * @param inputType the inputType where the aggregation operation will read from, must not be {@literal null} or * empty. - * @param outputType The parameterized type of the returned list, 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. * @since 1.3 */ @@ -416,7 +416,7 @@ public interface MongoOperations extends FluentMongoOperations { * {@literal null}. * @param collectionName the collection where the aggregation operation will read from, must not be {@literal null} or * empty. - * @param outputType The parameterized type of the returned list, 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. * @since 1.3 */ @@ -435,7 +435,7 @@ public interface MongoOperations extends FluentMongoOperations { * @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 parameterized type of the returned list, 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. * @since 2.0 */ @@ -453,7 +453,7 @@ public interface MongoOperations extends FluentMongoOperations { * * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be * {@literal null}. - * @param outputType The parameterized type of the returned list, 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. * @since 2.0 */ @@ -472,7 +472,7 @@ public interface MongoOperations extends FluentMongoOperations { * {@literal null}. * @param inputType the inputType where the aggregation operation will read from, must not be {@literal null} or * empty. - * @param outputType The parameterized type of the returned list, 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. * @since 2.0 */ @@ -491,7 +491,7 @@ public interface MongoOperations extends FluentMongoOperations { * {@literal null}. * @param collectionName the collection where the aggregation operation will read from, must not be {@literal null} or * empty. - * @param outputType The parameterized type of the returned list, 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. * @since 2.0 */ @@ -504,7 +504,7 @@ public interface MongoOperations extends FluentMongoOperations { * @param mapFunction The JavaScript map function * @param reduceFunction The JavaScript reduce function * @param mapReduceOptions Options that specify detailed map-reduce behavior - * @param entityClass The parameterized type of the returned list + * @param entityClass The parametrized type of the returned list * @return The results of the map reduce operation */ MapReduceResults mapReduce(String inputCollectionName, String mapFunction, String reduceFunction, @@ -517,7 +517,7 @@ public interface MongoOperations extends FluentMongoOperations { * @param mapFunction The JavaScript map function * @param reduceFunction The JavaScript reduce function * @param mapReduceOptions Options that specify detailed map-reduce behavior - * @param entityClass The parameterized type of the returned list + * @param entityClass The parametrized type of the returned list * @return The results of the map reduce operation */ MapReduceResults mapReduce(String inputCollectionName, String mapFunction, String reduceFunction, @@ -532,7 +532,7 @@ public interface MongoOperations extends FluentMongoOperations { * @param mapFunction The JavaScript map function * @param reduceFunction The JavaScript reduce function * @param mapReduceOptions Options that specify detailed map-reduce behavior - * @param entityClass The parameterized type of the returned list + * @param entityClass The parametrized type of the returned list * @return The results of the map reduce operation */ MapReduceResults mapReduce(Query query, String inputCollectionName, String mapFunction, String reduceFunction, @@ -546,7 +546,7 @@ public interface MongoOperations extends FluentMongoOperations { * @param mapFunction The JavaScript map function * @param reduceFunction The JavaScript reduce function * @param mapReduceOptions Options that specify detailed map-reduce behavior - * @param entityClass The parameterized type of the returned list + * @param entityClass The parametrized type of the returned list * @return The results of the map reduce operation */ MapReduceResults mapReduce(Query query, String inputCollectionName, String mapFunction, String reduceFunction, @@ -589,7 +589,7 @@ public interface MongoOperations extends FluentMongoOperations { * * @param query the query class that specifies the criteria used to find a record and also an optional fields * specification - * @param entityClass the parameterized type of the returned list. + * @param entityClass the parametrized type of the returned list. * @return the converted object */ T findOne(Query query, Class entityClass); @@ -606,7 +606,7 @@ public interface MongoOperations extends FluentMongoOperations { * * @param query the query class that specifies the criteria used to find a record and also an optional fields * specification - * @param entityClass the parameterized type of the returned list. + * @param entityClass the parametrized type of the returned list. * @param collectionName name of the collection to retrieve the objects from * @return the converted object */ @@ -627,7 +627,7 @@ public interface MongoOperations extends FluentMongoOperations { * Determine result of given {@link Query} contains at least one element. * * @param query the {@link Query} class that specifies the criteria used to find a record. - * @param entityClass the parameterized type. + * @param entityClass the parametrized type. * @return */ boolean exists(Query query, Class entityClass); @@ -636,7 +636,7 @@ public interface MongoOperations extends FluentMongoOperations { * Determine result of given {@link Query} contains at least one element. * * @param query the {@link Query} class that specifies the criteria used to find a record. - * @param entityClass the parameterized type. + * @param entityClass the parametrized type. * @param collectionName name of the collection to check for objects. * @return */ @@ -653,7 +653,7 @@ public interface MongoOperations extends FluentMongoOperations { * * @param query the query class that specifies the criteria used to find a record and also an optional fields * specification - * @param entityClass the parameterized type of the returned list. + * @param entityClass the parametrized type of the returned list. * @return the List of converted objects */ List find(Query query, Class entityClass); @@ -669,7 +669,7 @@ public interface MongoOperations extends FluentMongoOperations { * * @param query the query class that specifies the criteria used to find a record and also an optional fields * specification - * @param entityClass the parameterized type of the returned list. + * @param entityClass the parametrized type of the returned list. * @param collectionName name of the collection to retrieve the objects from * @return the List of converted objects */ @@ -704,7 +704,7 @@ public interface MongoOperations extends FluentMongoOperations { * @param query the {@link Query} class that specifies the {@link Criteria} used to find a record and also an optional * fields specification. * @param update the {@link Update} to apply on matching documents. - * @param entityClass the parameterized type. + * @param entityClass the parametrized type. * @return */ T findAndModify(Query query, Update update, Class entityClass); @@ -716,7 +716,7 @@ public interface MongoOperations extends FluentMongoOperations { * @param query the {@link Query} class that specifies the {@link Criteria} used to find a record and also an optional * fields specification. * @param update the {@link Update} to apply on matching documents. - * @param entityClass the parameterized type. + * @param entityClass the parametrized type. * @param collectionName the collection to query. * @return */ @@ -731,7 +731,7 @@ public interface MongoOperations extends FluentMongoOperations { * fields specification. * @param update the {@link Update} to apply on matching documents. * @param options the {@link FindAndModifyOptions} holding additional information. - * @param entityClass the parameterized type. + * @param entityClass the parametrized type. * @return */ T findAndModify(Query query, Update update, FindAndModifyOptions options, Class entityClass); @@ -745,7 +745,7 @@ public interface MongoOperations extends FluentMongoOperations { * fields specification. * @param update the {@link Update} to apply on matching documents. * @param options the {@link FindAndModifyOptions} holding additional information. - * @param entityClass the parameterized type. + * @param entityClass the parametrized type. * @param collectionName the collection to query. * @return */ @@ -764,7 +764,7 @@ public interface MongoOperations extends FluentMongoOperations { * * @param query the query class that specifies the criteria used to find a record and also an optional fields * specification - * @param entityClass the parameterized type of the returned list. + * @param entityClass the parametrized type of the returned list. * @return the converted object */ T findAndRemove(Query query, Class entityClass); @@ -781,7 +781,7 @@ public interface MongoOperations extends FluentMongoOperations { * * @param query the query class that specifies the criteria used to find a record and also an optional fields * specification - * @param entityClass the parameterized type of the returned list. + * @param entityClass the parametrized type of the returned list. * @param collectionName name of the collection to retrieve the objects from * @return the converted object */ diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoDbUtils.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoDbUtils.java deleted file mode 100644 index cb80c494a..000000000 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoDbUtils.java +++ /dev/null @@ -1,51 +0,0 @@ -/* - * Copyright 2016 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; - -import com.mongodb.reactivestreams.client.MongoClient; -import com.mongodb.reactivestreams.client.MongoDatabase; - -/** - * Helper class featuring helper methods for internal MongoDb classes. Mainly intended for internal use within the - * framework. - * - * @author Mark Paluch - * @author Christoph Strobl - * @since 2.0 - */ -public abstract class ReactiveMongoDbUtils { - - /** - * Private constructor to prevent instantiation. - */ - private ReactiveMongoDbUtils() {} - - /** - * Obtains a {@link MongoDatabase} connection for the given {@link MongoClient} instance and database name - * - * @param mongo the {@link MongoClient} instance, must not be {@literal null}. - * @param databaseName the database name, must not be {@literal null} or empty. - * @return the {@link MongoDatabase} connection - */ - public static MongoDatabase getMongoDatabase(MongoClient mongo, String databaseName) { - return doGetMongoDatabase(mongo, databaseName, true); - } - - private static MongoDatabase doGetMongoDatabase(MongoClient mongo, String databaseName, boolean allowCreate) { - return mongo.getDatabase(databaseName); - } - -} diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java index b0d6b32f4..6aac97f9b 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java @@ -376,17 +376,19 @@ public interface ReactiveMongoOperations { /** * Execute an aggregation operation. *

- * The raw results will be mapped to the given entity class. The name of the inputCollection is derived from the - * inputType of the aggregation. + * The raw results will be mapped to the given entity class. *

- * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling - * explanation mode will throw an {@link IllegalArgumentException}. + * Aggregation streaming cannot be used with {@link AggregationOptions#isExplain() aggregation explain} nor with + * {@link AggregationOptions#getCursorBatchSize()}. Enabling explanation mode or setting batch size cause + * {@link IllegalArgumentException}. * - * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be + * @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}. + * @param collectionName The name of the input collection to use for the aggregation. Must not be {@literal null}. + * @param outputType The parametrized type of the returned {@link Flux}. Must not be {@literal null}. * @return The results of the aggregation operation. + * @throws IllegalArgumentException if {@code aggregation}, {@code collectionName} or {@code outputType} is + * {@literal null}. */ Flux aggregate(TypedAggregation aggregation, String collectionName, Class outputType); @@ -394,32 +396,37 @@ public interface ReactiveMongoOperations { * Execute an aggregation operation. *

* 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. + * inputCollection is derived from the {@link TypedAggregation#getInputType() aggregation input type}. *

- * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling - * explanation mode will throw an {@link IllegalArgumentException}. + * Aggregation streaming cannot be used with {@link AggregationOptions#isExplain() aggregation explain} nor with + * {@link AggregationOptions#getCursorBatchSize()}. Enabling explanation mode or setting batch size cause + * {@link IllegalArgumentException}. * - * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be + * @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}. + * @param outputType The parametrized type of the returned {@link Flux}. Must not be {@literal null}. * @return The results of the aggregation operation. + * @throws IllegalArgumentException if {@code aggregation} or {@code outputType} is {@literal null}. */ Flux aggregate(TypedAggregation aggregation, Class outputType); /** * Execute an aggregation operation. *

- * The raw results will be mapped to the given entity class. + * The raw results will be mapped to the given {@code ouputType}. The name of the inputCollection is derived from the + * {@code inputType}. *

- * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling - * explanation mode will throw an {@link IllegalArgumentException}. + * Aggregation streaming cannot be used with {@link AggregationOptions#isExplain() aggregation explain} nor with + * {@link AggregationOptions#getCursorBatchSize()}. Enabling explanation mode or setting batch size cause + * {@link IllegalArgumentException}. * - * @param aggregation The {@link Aggregation} specification holding the aggregation operations, must not be + * @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}. + * @param inputType the inputType where the aggregation operation will read from. Must not be {@literal null}. + * @param outputType The parametrized type of the returned {@link Flux}. Must not be {@literal null}. * @return The results of the aggregation operation. + * @throws IllegalArgumentException if {@code aggregation}, {@code inputType} or {@code outputType} is + * {@literal null}. */ Flux aggregate(Aggregation aggregation, Class inputType, Class outputType); @@ -428,15 +435,18 @@ public interface ReactiveMongoOperations { *

* The raw results will be mapped to the given entity class. *

- * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling - * explanation mode will throw an {@link IllegalArgumentException}. + * Aggregation streaming cannot be used with {@link AggregationOptions#isExplain() aggregation explain} nor with + * {@link AggregationOptions#getCursorBatchSize()}. Enabling explanation mode or setting batch size cause + * {@link IllegalArgumentException}. * - * @param aggregation The {@link Aggregation} specification holding the aggregation operations, must not be + * @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 + * @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}. + * @param outputType The parametrized type of the returned {@link Flux}. Must not be {@literal null}. * @return The results of the aggregation operation. + * @throws IllegalArgumentException if {@code aggregation}, {@code collectionName} or {@code outputType} is + * {@literal null}. */ Flux aggregate(Aggregation aggregation, String collectionName, Class outputType); diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java index 8dec3d83e..10d591a3b 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java @@ -316,14 +316,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return this.mongoConverter; } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#reactiveIndexOps(java.lang.String) */ public ReactiveIndexOperations indexOps(String collectionName) { return new DefaultReactiveIndexOperations(this, collectionName, this.queryMapper); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#reactiveIndexOps(java.lang.Class) */ public ReactiveIndexOperations indexOps(Class entityClass) { @@ -335,7 +337,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return this.determineCollectionName(entityClass); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#executeCommand(java.lang.String) */ public Mono executeCommand(String jsonCommand) { @@ -345,14 +348,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return executeCommand(Document.parse(jsonCommand)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#executeCommand(org.bson.Document) */ public Mono executeCommand(final Document command) { return executeCommand(command, null); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#executeCommand(org.bson.Document, com.mongodb.ReadPreference) */ public Mono executeCommand(final Document command, final ReadPreference readPreference) { @@ -363,7 +368,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati : db.runCommand(command, Document.class)).next(); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#execute(java.lang.Class, org.springframework.data.mongodb.core.ReactiveCollectionCallback) */ @Override @@ -371,7 +377,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return createFlux(determineCollectionName(entityClass), action); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#execute(org.springframework.data.mongodb.core.ReactiveDbCallback) */ @Override @@ -379,7 +386,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return createFlux(action); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#execute(java.lang.String, org.springframework.data.mongodb.core.ReactiveCollectionCallback) */ public Flux execute(String collectionName, ReactiveCollectionCallback callback) { @@ -454,14 +462,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati .onErrorMap(translateException()); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#createCollection(java.lang.Class) */ public Mono> createCollection(Class entityClass) { return createCollection(determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#createCollection(java.lang.Class, org.springframework.data.mongodb.core.CollectionOptions) */ public Mono> createCollection(Class entityClass, @@ -469,14 +479,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return createCollection(determineCollectionName(entityClass), collectionOptions); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#createCollection(java.lang.String) */ public Mono> createCollection(final String collectionName) { return doCreateCollection(collectionName, new CreateCollectionOptions()); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#createCollection(java.lang.String, org.springframework.data.mongodb.core.CollectionOptions) */ public Mono> createCollection(final String collectionName, @@ -484,21 +496,24 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return doCreateCollection(collectionName, convertToCreateCollectionOptions(collectionOptions)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#getCollection(java.lang.String) */ public MongoCollection getCollection(final String collectionName) { return execute((MongoDatabaseCallback>) db -> db.getCollection(collectionName)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#collectionExists(java.lang.Class) */ public Mono collectionExists(Class entityClass) { return collectionExists(determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#collectionExists(java.lang.String) */ public Mono collectionExists(final String collectionName) { @@ -508,14 +523,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati .single(false)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#dropCollection(java.lang.Class) */ public Mono dropCollection(Class entityClass) { return dropCollection(determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#dropCollection(java.lang.String) */ public Mono dropCollection(final String collectionName) { @@ -527,7 +544,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati }).then(); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#getCollectionNames() */ public Flux getCollectionNames() { @@ -538,14 +556,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return mongoDatabaseFactory.getMongoDatabase(); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findOne(org.springframework.data.mongodb.core.query.Query, java.lang.Class) */ public Mono findOne(Query query, Class entityClass) { return findOne(query, entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findOne(org.springframework.data.mongodb.core.query.Query, java.lang.Class, java.lang.String) */ public Mono findOne(Query query, Class entityClass, String collectionName) { @@ -559,21 +579,24 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return find(query, entityClass, collectionName).next(); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#exists(org.springframework.data.mongodb.core.query.Query, java.lang.Class) */ public Mono exists(Query query, Class entityClass) { return exists(query, entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#exists(org.springframework.data.mongodb.core.query.Query, java.lang.String) */ public Mono exists(Query query, String collectionName) { return exists(query, null, collectionName); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#exists(org.springframework.data.mongodb.core.query.Query, java.lang.Class, java.lang.String) */ public Mono exists(final Query query, final Class entityClass, String collectionName) { @@ -594,14 +617,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati }).hasElements(); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#find(org.springframework.data.mongodb.core.query.Query, java.lang.Class) */ public Flux find(Query query, Class entityClass) { return find(query, entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#find(org.springframework.data.mongodb.core.query.Query, java.lang.Class, java.lang.String) */ public Flux find(final Query query, Class entityClass, String collectionName) { @@ -614,14 +639,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati new QueryFindPublisherPreparer(query, entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findById(java.lang.Object, java.lang.Class) */ public Mono findById(Object id, Class entityClass) { return findById(id, entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findById(java.lang.Object, java.lang.Class, java.lang.String) */ public Mono findById(Object id, Class entityClass, String collectionName) { @@ -634,7 +661,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return doFindOne(collectionName, new Document(idKey, id), null, entityClass, null); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.TypedAggregation, java.lang.String, java.lang.Class) */ @Override @@ -647,7 +675,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return aggregate(aggregation, inputCollectionName, outputType, context); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.TypedAggregation, java.lang.Class) */ @Override @@ -655,7 +684,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return aggregate(aggregation, determineCollectionName(aggregation.getInputType()), outputType); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.Aggregation, java.lang.Class, java.lang.Class) */ @Override @@ -665,7 +695,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati new TypeBasedAggregationOperationContext(inputType, mappingContext, queryMapper)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.Aggregation, java.lang.String, java.lang.Class) */ @Override @@ -673,43 +704,47 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return aggregate(aggregation, collectionName, outputType, null); } + /** + * @param aggregation must not be {@literal null}. + * @param collectionName must not be {@literal null}. + * @param outputType must not be {@literal null}. + * @param context can be {@literal null} and will be defaulted to {@link Aggregation#DEFAULT_CONTEXT}. + * @return never {@literal null}. + */ protected Flux aggregate(Aggregation aggregation, String collectionName, Class outputType, AggregationOperationContext context) { - Assert.hasText(collectionName, "Collection name must not be null or empty!"); Assert.notNull(aggregation, "Aggregation pipeline must not be null!"); + Assert.hasText(collectionName, "Collection name must not be null or empty!"); Assert.notNull(outputType, "Output type must not be null!"); AggregationOperationContext rootContext = context == null ? Aggregation.DEFAULT_CONTEXT : context; - Document command = aggregation.toDocument(collectionName, rootContext); + AggregationOptions options = AggregationOptions.fromDocument(command); - Boolean explain = command.get("explain", Boolean.class); - if (explain != null && explain) { - throw new IllegalArgumentException("Can't use explain option with streaming!"); - } + Assert.isTrue(!options.isExplain(), "Cannot use explain option with streaming!"); + Assert.isNull(options.getCursorBatchSize(), "Cannot use batchSize cursor option with streaming!"); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Streaming aggregation: {}", serializeToJsonSafely(command)); } ReadDocumentCallback readCallback = new ReadDocumentCallback<>(mongoConverter, outputType, collectionName); + return execute(collectionName, + collection -> aggregateAndMap(collection, (List) command.get("pipeline"), options, readCallback)); + } - return execute(collectionName, collection -> { + private Flux aggregateAndMap(MongoCollection collection, List pipeline, + AggregationOptions options, ReadDocumentCallback readCallback) { - List pipeline = (List) command.get("pipeline"); + AggregatePublisher cursor = collection.aggregate(pipeline).allowDiskUse(options.isAllowDiskUse()) + .useCursor(true); - AggregationOptions options = AggregationOptions.fromDocument(command); + if (options.getCollation().isPresent()) { + cursor = cursor.collation(options.getCollation().map(Collation::toMongoCollation).get()); + } - AggregatePublisher 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); - }); + return Flux.from(cursor).map(readCallback::doWith); } /* @@ -769,28 +804,32 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati }); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAndModify(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.Class) */ public Mono findAndModify(Query query, Update update, Class entityClass) { return findAndModify(query, update, new FindAndModifyOptions(), entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAndModify(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.Class, java.lang.String) */ public Mono findAndModify(Query query, Update update, Class entityClass, String collectionName) { return findAndModify(query, update, new FindAndModifyOptions(), entityClass, collectionName); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAndModify(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, org.springframework.data.mongodb.core.FindAndModifyOptions, java.lang.Class) */ public Mono findAndModify(Query query, Update update, FindAndModifyOptions options, Class entityClass) { return findAndModify(query, update, options, entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAndModify(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, org.springframework.data.mongodb.core.FindAndModifyOptions, java.lang.Class, java.lang.String) */ public Mono findAndModify(Query query, Update update, FindAndModifyOptions options, Class entityClass, @@ -809,14 +848,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati getMappedSortObject(query, entityClass), entityClass, update, optionsToUse); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAndRemove(org.springframework.data.mongodb.core.query.Query, java.lang.Class) */ public Mono findAndRemove(Query query, Class entityClass) { return findAndRemove(query, entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAndRemove(org.springframework.data.mongodb.core.query.Query, java.lang.Class, java.lang.String) */ public Mono findAndRemove(Query query, Class entityClass, String collectionName) { @@ -825,7 +866,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati getMappedSortObject(query, entityClass), query.getCollation().orElse(null), entityClass); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#count(org.springframework.data.mongodb.core.query.Query, java.lang.Class) */ public Mono count(Query query, Class entityClass) { @@ -835,14 +877,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return count(query, entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#count(org.springframework.data.mongodb.core.query.Query, java.lang.String) */ public Mono count(final Query query, String collectionName) { return count(query, null, collectionName); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#count(org.springframework.data.mongodb.core.query.Query, java.lang.Class, java.lang.String) */ public Mono count(final Query query, final Class entityClass, String collectionName) { @@ -859,7 +903,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati }); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#insert(reactor.core.publisher.Mono) */ @Override @@ -867,7 +912,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return objectToSave.flatMap(this::insert); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#insert(org.reactivestreams.Publisher, java.lang.Class) */ @Override @@ -875,7 +921,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return insertAll(batchToSave, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#insert(org.reactivestreams.Publisher, java.lang.String) */ @Override @@ -883,7 +930,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return Flux.from(batchToSave).flatMap(collection -> insert(collection, collectionName)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#insert(java.lang.Object) */ public Mono insert(T objectToSave) { @@ -892,7 +940,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return insert(objectToSave, determineEntityCollectionName(objectToSave)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#insert(java.lang.Object, java.lang.String) */ public Mono insert(T objectToSave, String collectionName) { @@ -924,28 +973,32 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati }); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#insert(java.util.Collection, java.lang.Class) */ public Flux insert(Collection batchToSave, Class entityClass) { return doInsertBatch(determineCollectionName(entityClass), batchToSave, this.mongoConverter); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#insert(java.util.Collection, java.lang.String) */ public Flux insert(Collection batchToSave, String collectionName) { return doInsertBatch(collectionName, batchToSave, this.mongoConverter); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#insertAll(java.util.Collection) */ public Flux insertAll(Collection objectsToSave) { return doInsertAll(objectsToSave, this.mongoConverter); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#insertAll(org.reactivestreams.Publisher) */ @Override @@ -1010,7 +1063,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati }); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#save(reactor.core.publisher.Mono) */ @Override @@ -1018,7 +1072,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return objectToSave.flatMap(this::save); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#save(reactor.core.publisher.Mono, java.lang.String) */ @Override @@ -1026,7 +1081,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return objectToSave.flatMap(o -> save(o, collectionName)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#save(java.lang.Object) */ public Mono save(T objectToSave) { @@ -1035,7 +1091,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return save(objectToSave, determineEntityCollectionName(objectToSave)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#save(java.lang.Object, java.lang.String) */ public Mono save(T objectToSave, String collectionName) { @@ -1210,63 +1267,72 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati }); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#upsert(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.Class) */ public Mono upsert(Query query, Update update, Class entityClass) { return doUpdate(determineCollectionName(entityClass), query, update, entityClass, true, false); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#upsert(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.String) */ public Mono upsert(Query query, Update update, String collectionName) { return doUpdate(collectionName, query, update, null, true, false); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#upsert(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.Class, java.lang.String) */ public Mono upsert(Query query, Update update, Class entityClass, String collectionName) { return doUpdate(collectionName, query, update, entityClass, true, false); } - /* (non-Javadoc) + /* + * (non-Javadoc)) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#updateFirst(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.Class) */ public Mono updateFirst(Query query, Update update, Class entityClass) { return doUpdate(determineCollectionName(entityClass), query, update, entityClass, false, false); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#updateFirst(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.String) */ public Mono updateFirst(final Query query, final Update update, final String collectionName) { return doUpdate(collectionName, query, update, null, false, false); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#updateFirst(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.Class, java.lang.String) */ public Mono updateFirst(Query query, Update update, Class entityClass, String collectionName) { return doUpdate(collectionName, query, update, entityClass, false, false); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#updateMulti(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.Class) */ public Mono updateMulti(Query query, Update update, Class entityClass) { return doUpdate(determineCollectionName(entityClass), query, update, entityClass, false, true); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#updateMulti(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.String) */ public Mono updateMulti(final Query query, final Update update, String collectionName) { return doUpdate(collectionName, query, update, null, false, true); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#updateMulti(org.springframework.data.mongodb.core.query.Query, org.springframework.data.mongodb.core.query.Update, java.lang.Class, java.lang.String) */ public Mono updateMulti(final Query query, final Update update, Class entityClass, @@ -1345,7 +1411,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return document.containsKey(persistentEntity.getRequiredIdProperty().getFieldName()); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#remove(reactor.core.publisher.Mono) */ @Override @@ -1353,7 +1420,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return objectToRemove.flatMap(this::remove); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#remove(reactor.core.publisher.Mono, java.lang.String) */ @Override @@ -1361,7 +1429,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return objectToRemove.flatMap(o -> remove(objectToRemove, collection)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#remove(java.lang.Object) */ public Mono remove(Object object) { @@ -1373,7 +1442,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return remove(getIdQueryFor(object), object.getClass()); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#remove(java.lang.Object, java.lang.String) */ public Mono remove(Object object, String collection) { @@ -1473,21 +1543,24 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati } } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#remove(org.springframework.data.mongodb.core.query.Query, java.lang.String) */ public Mono remove(Query query, String collectionName) { return remove(query, null, collectionName); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#remove(org.springframework.data.mongodb.core.query.Query, java.lang.Class) */ public Mono remove(Query query, Class entityClass) { return remove(query, entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#remove(org.springframework.data.mongodb.core.query.Query, java.lang.Class, java.lang.String) */ public Mono remove(Query query, Class entityClass, String collectionName) { @@ -1534,14 +1607,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati .next(); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAll(java.lang.Class) */ public Flux findAll(Class entityClass) { return findAll(entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAll(java.lang.Class, java.lang.String) */ public Flux findAll(Class entityClass, String collectionName) { @@ -1549,7 +1624,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati new ReadDocumentCallback(mongoConverter, entityClass, collectionName), collectionName); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAllAndRemove(org.springframework.data.mongodb.core.query.Query, java.lang.String) */ @Override @@ -1558,7 +1634,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return findAllAndRemove(query, null, collectionName); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAllAndRemove(org.springframework.data.mongodb.core.query.Query, java.lang.Class) */ @Override @@ -1566,7 +1643,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return findAllAndRemove(query, entityClass, determineCollectionName(entityClass)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#findAllAndRemove(org.springframework.data.mongodb.core.query.Query, java.lang.Class, java.lang.String) */ @Override @@ -1574,11 +1652,19 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return doFindAndDelete(collectionName, query, entityClass); } + /* + * (non-Javadoc) + * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#tail(org.springframework.data.mongodb.core.query.Query, java.lang.Class) + */ @Override public Flux tail(Query query, Class entityClass) { return tail(query, entityClass, determineCollectionName(entityClass)); } + /* + * (non-Javadoc) + * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#tail(org.springframework.data.mongodb.core.query.Query, java.lang.Class, java.lang.String) + */ @Override public Flux tail(Query query, Class entityClass, String collectionName) { diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java index 78431860b..325983fc7 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/SimpleReactiveMongoDatabaseFactory.java @@ -33,6 +33,7 @@ import com.mongodb.reactivestreams.client.MongoDatabase; * Factory to create {@link MongoDatabase} instances from a {@link MongoClient} instance. * * @author Mark Paluch + * @author Christoph Strobl * @since 2.0 */ public class SimpleReactiveMongoDatabaseFactory implements DisposableBean, ReactiveMongoDatabaseFactory { @@ -103,14 +104,8 @@ public class SimpleReactiveMongoDatabaseFactory implements DisposableBean, React Assert.hasText(dbName, "Database name must not be empty."); - MongoDatabase db = ReactiveMongoDbUtils.getMongoDatabase(mongo, dbName); - - if (writeConcern != null) { - - db = db.withWriteConcern(writeConcern); - } - - return db; + MongoDatabase db = mongo.getDatabase(dbName); + return writeConcern != null ? db.withWriteConcern(writeConcern) : db; } /** @@ -119,6 +114,7 @@ public class SimpleReactiveMongoDatabaseFactory implements DisposableBean, React * @see DisposableBean#destroy() */ public void destroy() throws Exception { + if (mongoInstanceCreated) { mongo.close(); } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationOptions.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationOptions.java index d88f3f5c3..694aeb108 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationOptions.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationOptions.java @@ -110,6 +110,16 @@ public class AggregationOptions { return new AggregationOptions(allowDiskUse, explain, cursor, collation); } + /** + * Obtain a new {@link Builder} for constructing {@link AggregationOptions}. + * + * @return never {@literal null}. + * @since 2.0 + */ + public static Builder builder() { + return new Builder(); + } + /** * Enables writing to temporary files. When set to true, aggregation stages can write data to the _tmp subdirectory in * the dbPath directory. diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationUnitTests.java index 5929735d4..ba36ee760 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationUnitTests.java @@ -382,7 +382,6 @@ public class AggregationUnitTests { .as("foosum") // ).toDocument("foo", Aggregation.DEFAULT_CONTEXT); - System.out.println("agg: " + agg); @SuppressWarnings("unchecked") Document secondProjection = ((List) agg.get("pipeline")).get(1); Document fields = getAsDocument(secondProjection, "$group"); diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationTests.java index b94e9937a..af4834aa2 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationTests.java @@ -38,6 +38,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * Integration test for aggregation via {@link org.springframework.data.mongodb.core.ReactiveMongoTemplate}. * * @author Mark Paluch + * @author Christoph Strobl */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration("classpath:reactive-infrastructure.xml") @@ -59,6 +60,7 @@ public class ReactiveAggregationTests { } private void cleanDb() { + StepVerifier .create(reactiveMongoTemplate.dropCollection(INPUT_COLLECTION) // .then(reactiveMongoTemplate.dropCollection(OUTPUT_COLLECTION)) // @@ -67,21 +69,6 @@ public class ReactiveAggregationTests { .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() { diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationUnitTests.java new file mode 100644 index 000000000..4dce1d056 --- /dev/null +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationUnitTests.java @@ -0,0 +1,141 @@ +/* + * 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.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.*; +import static org.springframework.data.mongodb.core.aggregation.Aggregation.*; + +import org.bson.Document; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; +import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; +import org.springframework.data.mongodb.core.ReactiveMongoTemplate; +import org.springframework.data.mongodb.core.SimpleReactiveMongoDatabaseFactory; +import org.springframework.data.mongodb.core.query.Collation; + +import com.mongodb.reactivestreams.client.AggregatePublisher; +import com.mongodb.reactivestreams.client.MongoClient; +import com.mongodb.reactivestreams.client.MongoCollection; +import com.mongodb.reactivestreams.client.MongoDatabase; + +/** + * @author Christoph Strobl + * @author Mark Paluch + */ +@RunWith(MockitoJUnitRunner.class) +public class ReactiveAggregationUnitTests { + + static final String INPUT_COLLECTION = "collection-1"; + + ReactiveMongoTemplate template; + ReactiveMongoDatabaseFactory factory; + @Mock MongoClient mongoClient; + @Mock MongoDatabase db; + @Mock MongoCollection collection; + @Mock AggregatePublisher publisher; + + @Before + public void setUp() { + + factory = new SimpleReactiveMongoDatabaseFactory(mongoClient, "db"); + template = new ReactiveMongoTemplate(factory); + + when(mongoClient.getDatabase("db")).thenReturn(db); + when(db.getCollection(INPUT_COLLECTION)).thenReturn(collection); + when(collection.aggregate(any())).thenReturn(publisher); + when(publisher.allowDiskUse(any())).thenReturn(publisher); + when(publisher.useCursor(any())).thenReturn(publisher); + when(publisher.collation(any())).thenReturn(publisher); + } + + @Test(expected = IllegalArgumentException.class) // DATAMONGO-1646 + public void shouldHandleMissingInputCollection() { + template.aggregate(newAggregation(), (String) null, TagCount.class); + } + + @Test(expected = IllegalArgumentException.class) // DATAMONGO-1646 + public void shouldHandleMissingAggregationPipeline() { + template.aggregate(null, INPUT_COLLECTION, TagCount.class); + } + + @Test(expected = IllegalArgumentException.class) // DATAMONGO-1646 + public void shouldHandleMissingEntityClass() { + template.aggregate(newAggregation(), INPUT_COLLECTION, null); + } + + @Test(expected = IllegalArgumentException.class) // DATAMONGO-1646 + public void errorsOnCursorBatchSizeUsage() { + + template.aggregate( + newAggregation(Product.class, // + project("name", "netPrice")) // + .withOptions(AggregationOptions.builder().cursorBatchSize(10).build()), + INPUT_COLLECTION, TagCount.class).subscribe(); + } + + @Test(expected = IllegalArgumentException.class) // DATAMONGO-1646 + public void errorsOnExplainUsage() { + + template + .aggregate(newAggregation(Product.class, // + project("name", "netPrice")) // + .withOptions(AggregationOptions.builder().explain(true).build()), + INPUT_COLLECTION, TagCount.class) + .subscribe(); + } + + @Test // DATAMONGO-1646 + public void appliesCollationCorrectlyWhenPresent() { + + template.aggregate( + newAggregation(Product.class, // + project("name", "netPrice")) // + .withOptions(AggregationOptions.builder().collation(Collation.of("en_US")).build()), + INPUT_COLLECTION, TagCount.class).subscribe(); + + verify(publisher).collation(eq(com.mongodb.client.model.Collation.builder().locale("en_US").build())); + } + + @Test // DATAMONGO-1646 + public void doesNotSetCollationWhenNotPresent() { + + template.aggregate(newAggregation(Product.class, // + project("name", "netPrice")) // + .withOptions(AggregationOptions.builder().build()), + INPUT_COLLECTION, TagCount.class).subscribe(); + + verify(publisher, never()).collation(any()); + } + + @Test // DATAMONGO-1646 + public void appliesDiskUsageCorrectly() { + + template + .aggregate( + newAggregation(Product.class, // + project("name", "netPrice")) // + .withOptions(AggregationOptions.builder().allowDiskUse(true).build()), + INPUT_COLLECTION, TagCount.class) + .subscribe(); + + verify(publisher).allowDiskUse(eq(true)); + } +}