Skip to content

Commit 06622be

Browse files
mp911demp911de
committed
DATAMONGO-1986 - Polishing.
Refactor duplicated code into AggregationUtil. Original pull request: #564.
1 parent 2bac54c commit 06622be

File tree

3 files changed

+126
-114
lines changed

3 files changed

+126
-114
lines changed

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

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
/*
2+
* Copyright 2018 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package org.springframework.data.mongodb.core;
17+
18+
import lombok.AllArgsConstructor;
19+
20+
import java.util.List;
21+
import java.util.Optional;
22+
import java.util.stream.Collectors;
23+
24+
import org.bson.Document;
25+
import org.springframework.data.mapping.context.MappingContext;
26+
import org.springframework.data.mongodb.core.aggregation.Aggregation;
27+
import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
28+
import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
29+
import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
30+
import org.springframework.data.mongodb.core.convert.QueryMapper;
31+
import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
32+
import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
33+
import org.springframework.lang.Nullable;
34+
import org.springframework.util.ObjectUtils;
35+
36+
/**
37+
* Utility methods to map {@link org.springframework.data.mongodb.core.aggregation.Aggregation} pipeline definitions and
38+
* create type-bound {@link AggregationOperationContext}.
39+
*
40+
* @author Christoph Strobl
41+
* @author Mark Paluch
42+
* @since 2.1
43+
*/
44+
@AllArgsConstructor
45+
class AggregationUtil {
46+
47+
QueryMapper queryMapper;
48+
MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> mappingContext;
49+
50+
/**
51+
* Prepare the {@link AggregationOperationContext} for a given aggregation by either returning the context itself it
52+
* is not {@literal null}, create a {@link TypeBasedAggregationOperationContext} if the aggregation contains type
53+
* information (is a {@link TypedAggregation}) or use the {@link Aggregation#DEFAULT_CONTEXT}.
54+
*
55+
* @param aggregation must not be {@literal null}.
56+
* @param context can be {@literal null}.
57+
* @return the root {@link AggregationOperationContext} to use.
58+
*/
59+
AggregationOperationContext prepareAggregationContext(Aggregation aggregation,
60+
@Nullable AggregationOperationContext context) {
61+
62+
if (context != null) {
63+
return context;
64+
}
65+
66+
if (aggregation instanceof TypedAggregation) {
67+
return new TypeBasedAggregationOperationContext(((TypedAggregation) aggregation).getInputType(), mappingContext,
68+
queryMapper);
69+
}
70+
71+
return Aggregation.DEFAULT_CONTEXT;
72+
}
73+
74+
/**
75+
* Extract and map the aggregation pipeline into a {@link List} of {@link Document}.
76+
*
77+
* @param aggregation
78+
* @param context
79+
* @return
80+
*/
81+
List<Document> createPipeline(Aggregation aggregation, AggregationOperationContext context) {
82+
83+
if (!ObjectUtils.nullSafeEquals(context, Aggregation.DEFAULT_CONTEXT)) {
84+
return aggregation.toPipeline(context);
85+
}
86+
87+
return mapAggregationPipeline(aggregation.toPipeline(context));
88+
}
89+
90+
/**
91+
* Extract the command and map the aggregation pipeline.
92+
*
93+
* @param aggregation
94+
* @param context
95+
* @return
96+
*/
97+
Document createCommand(String collection, Aggregation aggregation, AggregationOperationContext context) {
98+
99+
Document command = aggregation.toDocument(collection, context);
100+
101+
if (!ObjectUtils.nullSafeEquals(context, Aggregation.DEFAULT_CONTEXT)) {
102+
return command;
103+
}
104+
105+
command.put("pipeline", mapAggregationPipeline(command.get("pipeline", List.class)));
106+
107+
return command;
108+
}
109+
110+
private List<Document> mapAggregationPipeline(List<Document> pipeline) {
111+
112+
return pipeline.stream().map(val -> queryMapper.getMappedObject(val, Optional.empty()))
113+
.collect(Collectors.toList());
114+
}
115+
}

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

Lines changed: 8 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -2116,7 +2116,8 @@ protected <O> AggregationResults<O> aggregate(Aggregation aggregation, String co
21162116
Assert.notNull(aggregation, "Aggregation pipeline must not be null!");
21172117
Assert.notNull(outputType, "Output type must not be null!");
21182118

2119-
return doAggregate(aggregation, collectionName, outputType, prepareAggregationContext(aggregation, context));
2119+
AggregationOperationContext contextToUse = new AggregationUtil(queryMapper, mappingContext).prepareAggregationContext(aggregation, context);
2120+
return doAggregate(aggregation, collectionName, outputType, contextToUse);
21202121
}
21212122

21222123
@SuppressWarnings("ConstantConditions")
@@ -2126,10 +2127,11 @@ protected <O> AggregationResults<O> doAggregate(Aggregation aggregation, String
21262127
DocumentCallback<O> callback = new UnwrapAndReadDocumentCallback<>(mongoConverter, outputType, collectionName);
21272128

21282129
AggregationOptions options = aggregation.getOptions();
2130+
AggregationUtil aggregationUtil = new AggregationUtil(queryMapper, mappingContext);
21292131

21302132
if (options.isExplain()) {
21312133

2132-
Document command = aggregationToCommand(collectionName, aggregation, context);
2134+
Document command = aggregationUtil.createCommand(collectionName, aggregation, context);
21332135

21342136
if (LOGGER.isDebugEnabled()) {
21352137
LOGGER.debug("Executing aggregation: {}", serializeToJsonSafely(command));
@@ -2140,7 +2142,7 @@ protected <O> AggregationResults<O> doAggregate(Aggregation aggregation, String
21402142
.map(callback::doWith).collect(Collectors.toList()), commandResult);
21412143
}
21422144

2143-
List<Document> pipeline = aggregationToPipeline(aggregation, context);
2145+
List<Document> pipeline = aggregationUtil.createPipeline(aggregation, context);
21442146

21452147
if (LOGGER.isDebugEnabled()) {
21462148
LOGGER.debug("Executing aggregation: {} in collection {}", serializeToJsonSafely(pipeline), collectionName);
@@ -2178,10 +2180,11 @@ protected <O> CloseableIterator<O> aggregateStream(Aggregation aggregation, Stri
21782180
Assert.notNull(outputType, "Output type must not be null!");
21792181
Assert.isTrue(!aggregation.getOptions().isExplain(), "Can't use explain option with streaming!");
21802182

2181-
AggregationOperationContext rootContext = prepareAggregationContext(aggregation, context);
2183+
AggregationUtil aggregationUtil = new AggregationUtil(queryMapper, mappingContext);
2184+
AggregationOperationContext rootContext = aggregationUtil.prepareAggregationContext(aggregation, context);
21822185

21832186
AggregationOptions options = aggregation.getOptions();
2184-
List<Document> pipeline = aggregationToPipeline(aggregation, rootContext);
2187+
List<Document> pipeline = aggregationUtil.createPipeline(aggregation, rootContext);
21852188

21862189
if (LOGGER.isDebugEnabled()) {
21872190
LOGGER.debug("Streaming aggregation: {} in collection {}", serializeToJsonSafely(pipeline), collectionName);
@@ -2844,72 +2847,6 @@ private Document addFieldsForProjection(Document fields, Class<?> domainType, Cl
28442847
return fields;
28452848
}
28462849

2847-
/**
2848-
* Prepare the {@link AggregationOperationContext} for a given aggregation by either returning the context itself it
2849-
* is not {@literal null}, create a {@link TypeBasedAggregationOperationContext} if the aggregation contains type
2850-
* information (is a {@link TypedAggregation}) or use the {@link Aggregation#DEFAULT_CONTEXT}.
2851-
*
2852-
* @param aggregation must not be {@literal null}.
2853-
* @param context can be {@literal null}.
2854-
* @return the root {@link AggregationOperationContext} to use.
2855-
*/
2856-
private AggregationOperationContext prepareAggregationContext(Aggregation aggregation,
2857-
@Nullable AggregationOperationContext context) {
2858-
2859-
if (context != null) {
2860-
return context;
2861-
}
2862-
2863-
if (aggregation instanceof TypedAggregation) {
2864-
return new TypeBasedAggregationOperationContext(((TypedAggregation) aggregation).getInputType(), mappingContext,
2865-
queryMapper);
2866-
}
2867-
2868-
return Aggregation.DEFAULT_CONTEXT;
2869-
}
2870-
2871-
/**
2872-
* Extract and map the aggregation pipeline.
2873-
*
2874-
* @param aggregation
2875-
* @param context
2876-
* @return
2877-
*/
2878-
private List<Document> aggregationToPipeline(Aggregation aggregation, AggregationOperationContext context) {
2879-
2880-
if (!ObjectUtils.nullSafeEquals(context, Aggregation.DEFAULT_CONTEXT)) {
2881-
return aggregation.toPipeline(context);
2882-
}
2883-
2884-
return mapAggregationPipeline(aggregation.toPipeline(context));
2885-
}
2886-
2887-
/**
2888-
* Extract the command and map the aggregation pipeline.
2889-
*
2890-
* @param aggregation
2891-
* @param context
2892-
* @return
2893-
*/
2894-
private Document aggregationToCommand(String collection, Aggregation aggregation,
2895-
AggregationOperationContext context) {
2896-
2897-
Document command = aggregation.toDocument(collection, context);
2898-
2899-
if (!ObjectUtils.nullSafeEquals(context, Aggregation.DEFAULT_CONTEXT)) {
2900-
return command;
2901-
}
2902-
2903-
command.put("pipeline", mapAggregationPipeline(command.get("pipeline", List.class)));
2904-
2905-
return command;
2906-
}
2907-
2908-
private List<Document> mapAggregationPipeline(List<Document> pipeline) {
2909-
2910-
return pipeline.stream().map(val -> queryMapper.getMappedObject(val, Optional.empty()))
2911-
.collect(Collectors.toList());
2912-
}
29132850

29142851
/**
29152852
* Tries to convert the given {@link RuntimeException} into a {@link DataAccessException} but returns the original

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

Lines changed: 3 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -969,10 +969,11 @@ protected <O> Flux<O> aggregate(Aggregation aggregation, String collectionName,
969969
Assert.hasText(collectionName, "Collection name must not be null or empty!");
970970
Assert.notNull(outputType, "Output type must not be null!");
971971

972-
AggregationOperationContext rootContext = prepareAggregationContext(aggregation, context);
972+
AggregationUtil aggregationUtil = new AggregationUtil(queryMapper, mappingContext);
973+
AggregationOperationContext rootContext = aggregationUtil.prepareAggregationContext(aggregation, context);
973974

974975
AggregationOptions options = aggregation.getOptions();
975-
List<Document> pipeline = aggregationToPipeline(aggregation, rootContext);
976+
List<Document> pipeline = aggregationUtil.createPipeline(aggregation, rootContext);
976977

977978
Assert.isTrue(!options.isExplain(), "Cannot use explain option with streaming!");
978979
Assert.isNull(options.getCursorBatchSize(), "Cannot use batchSize cursor option with streaming!");
@@ -2654,47 +2655,6 @@ private Function<Throwable, Throwable> translateException() {
26542655
};
26552656
}
26562657

2657-
/**
2658-
* Prepare the {@link AggregationOperationContext} for a given aggregation by either returning the context itself it
2659-
* is not {@literal null}, create a {@link TypeBasedAggregationOperationContext} if the aggregation contains type
2660-
* information (is a {@link TypedAggregation}) or use the {@link Aggregation#DEFAULT_CONTEXT}.
2661-
*
2662-
* @param aggregation must not be {@literal null}.
2663-
* @param context can be {@literal null}.
2664-
* @return the root {@link AggregationOperationContext} to use.
2665-
*/
2666-
private AggregationOperationContext prepareAggregationContext(Aggregation aggregation,
2667-
@Nullable AggregationOperationContext context) {
2668-
2669-
if (context != null) {
2670-
return context;
2671-
}
2672-
2673-
if (aggregation instanceof TypedAggregation) {
2674-
return new TypeBasedAggregationOperationContext(((TypedAggregation) aggregation).getInputType(), mappingContext,
2675-
queryMapper);
2676-
}
2677-
2678-
return Aggregation.DEFAULT_CONTEXT;
2679-
}
2680-
2681-
/**
2682-
* Extract and map the aggregation pipeline.
2683-
*
2684-
* @param aggregation
2685-
* @param context
2686-
* @return
2687-
*/
2688-
private List<Document> aggregationToPipeline(Aggregation aggregation, AggregationOperationContext context) {
2689-
2690-
if (!ObjectUtils.nullSafeEquals(context, Aggregation.DEFAULT_CONTEXT)) {
2691-
return aggregation.toPipeline(context);
2692-
}
2693-
2694-
return aggregation.toPipeline(context).stream().map(val -> queryMapper.getMappedObject(val, Optional.empty()))
2695-
.collect(Collectors.toList());
2696-
}
2697-
26982658
/**
26992659
* Tries to convert the given {@link RuntimeException} into a {@link DataAccessException} but returns the original
27002660
* exception if the conversation failed. Thus allows safe re-throwing of the return value.

0 commit comments

Comments
 (0)