Add support for $expMovingAvg aggregation operator.

The SpEL support for this one is missing due to the differing argument map (N, alpha).

Closes: #3718

Author: christophstrobl

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

@@ -142,6 +142,36 @@ public StdDevSamp stdDevSamp() {
 			return usesFieldRef() ? StdDevSamp.stdDevSampOf(fieldReference) : StdDevSamp.stdDevSampOf(expression);
 		}
 
+		/**
+		 * Creates new {@link AggregationExpression} that calculates the exponential moving average of numeric values
+		 * considering the given number of historical documents with significant weight.
+		 *
+		 * @param numberOfHistoricalDocuments the number of historical documents.
+		 * @return new instance of {@link ExpMovingAvg}.
+		 * @since 3.3
+		 */
+		public ExpMovingAvg expMovingAvg(int numberOfHistoricalDocuments) {
+
+			ExpMovingAvg expMovingAvg = usesFieldRef() ? ExpMovingAvg.expMovingAvgOf(fieldReference)
+					: ExpMovingAvg.expMovingAvgOf(expression);
+			return expMovingAvg.N(numberOfHistoricalDocuments);
+		}
+
+		/**
+		 * Creates new {@link AggregationExpression} that calculates the exponential moving average of numeric values
+		 * applying the given exponential decay value.
+		 *
+		 * @param exponentialDecayValue the decay value.
+		 * @return new instance of {@link ExpMovingAvg}.
+		 * @since 3.3
+		 */
+		public ExpMovingAvg expMovingAvg(double exponentialDecayValue) {
+
+			ExpMovingAvg expMovingAvg = usesFieldRef() ? ExpMovingAvg.expMovingAvgOf(fieldReference)
+					: ExpMovingAvg.expMovingAvgOf(expression);
+			return expMovingAvg.alpha(exponentialDecayValue);
+		}
+
 		private boolean usesFieldRef() {
 			return fieldReference != null;
 		}
@@ -658,4 +688,65 @@ public Document toDocument(Object value, AggregationOperationContext context) {
 			return super.toDocument(value, context);
 		}
 	}
+
+	/**
+	 * {@link ExpMovingAvg} calculates the exponential moving average of numeric values.
+	 *
+	 * @author Christoph Strobl
+	 * @since 3.3
+	 */
+	public static class ExpMovingAvg extends AbstractAggregationExpression {
+
+		private ExpMovingAvg(Object value) {
+			super(value);
+		}
+
+		/**
+		 * Create a new {@link ExpMovingAvg} by defining the field holding the value to be used as input.
+		 *
+		 * @param fieldReference must not be {@literal null}.
+		 * @return new instance of {@link ExpMovingAvg}.
+		 */
+		public static ExpMovingAvg expMovingAvgOf(String fieldReference) {
+			return new ExpMovingAvg(Collections.singletonMap("input", Fields.field(fieldReference)));
+		}
+
+		/**
+		 * Create a new {@link ExpMovingAvg} by defining the {@link AggregationExpression expression} to compute the value
+		 * to be used as input.
+		 *
+		 * @param expression must not be {@literal null}.
+		 * @return new instance of {@link ExpMovingAvg}.
+		 */
+		public static ExpMovingAvg expMovingAvgOf(AggregationExpression expression) {
+			return new ExpMovingAvg(Collections.singletonMap("input", expression));
+		}
+
+		/**
+		 * Define the number of historical documents with significant mathematical weight. <br />
+		 * Specify either {@link #N(int) N} or {@link #alpha(double) aplha}. Not both!
+		 *
+		 * @param numberOfHistoricalDocuments
+		 * @return new instance of {@link ExpMovingAvg}.
+		 */
+		public ExpMovingAvg N/*umber of historical documents*/(int numberOfHistoricalDocuments) {
+			return new ExpMovingAvg(append("N", numberOfHistoricalDocuments));
+		}
+
+		/**
+		 * Define the exponential decay value. <br />
+		 * Specify either {@link #alpha(double) aplha} or {@link #N(int) N}. Not both!
+		 *
+		 * @param exponentialDecayValue
+		 * @return new instance of {@link ExpMovingAvg}.
+		 */
+		public ExpMovingAvg alpha(double exponentialDecayValue) {
+			return new ExpMovingAvg(append("alpha", exponentialDecayValue));
+		}
+
+		@Override
+		protected String getMongoMethod() {
+			return "$expMovingAvg";
+		}
+	}
 }

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2021 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
+ *
+ *      https://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.Assertions.*;
+import static org.springframework.data.mongodb.core.aggregation.AccumulatorOperators.*;
+
+import org.bson.Document;
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author Christoph Strobl
+ */
+class AccumulatorOperatorsUnitTests {
+
+	@Test // GH-3718
+	void rendersExpMovingAvgWithN/*umberOfHistoricDocuments*/() {
+
+		assertThat(valueOf("price").expMovingAvg(2).toDocument(Aggregation.DEFAULT_CONTEXT))
+				.isEqualTo(Document.parse("{ $expMovingAvg: { input: \"$price\", N: 2 } }"));
+	}
+
+	@Test // GH-3718
+	void rendersExpMovingAvgWithAlpha() {
+
+		assertThat(valueOf("price").expMovingAvg(0.75).toDocument(Aggregation.DEFAULT_CONTEXT))
+				.isEqualTo(Document.parse("{ $expMovingAvg: { input: \"$price\", alpha: 0.75 } }"));
+	}
+
+}