Improve averageBigInteger and sumBigInteger precision/conversion (#286)

This commit improves the precision and conversion behavior of `MathFlux`
`sumBigInteger` and `averageBigInteger` methods. These method better deal with
fractional types by internally summing using a `BigDecimal` rather than a
`BigInteger`. As a result, the only lossy step is at the end when the result is
effectively presented as a `BigInteger`: 
 - `sum`: remaining fractional part of the BigDecimal sum is dropped
 - `average`: the division by `count` is applied, then the result's fractional
 part is dropped

Previously the fractional part would be progressively dropped by implicit
conversion of each incoming element to `long`.

Similar to precision improvement done in #260/#261 and earlier iteration done
in #284.

reactor-extra/src/main/java/reactor/math/MathFlux.java

@@ -143,6 +143,10 @@ public static final <T> Mono<Double> sumDouble(Publisher<T> source, Function<? s
 
 	/**
 	 * Computes the {@link BigInteger} sum of items in the source.
+	 * <p>
+	 * Further conversion of individual elements to {@link BigDecimal} is always applied to
+	 * retain maximum precision. When the sum is ultimately produced, it can be rounded or
+	 * truncated as a {@link BigDecimal#toBigInteger()} final conversion is applied.
 	 *
 	 * @param source the numerical source
 	 * @return {@link Mono} of the sum of items in source
@@ -154,6 +158,10 @@ public static Mono<BigInteger> sumBigInteger(Publisher<? extends Number> source)
 	/**
 	 * Computes the {@link BigInteger} sum of items in the source, which are mapped to
 	 * numerical values using provided mapping.
+	 * <p>
+	 * Further conversion of individual elements to {@link BigDecimal} is always applied to
+	 * retain maximum precision. When the sum is ultimately produced, it can be rounded or
+	 * truncated as a {@link BigDecimal#toBigInteger()} final conversion is applied.
 	 *
 	 * @param source  the source items
 	 * @param mapping a function to map source items to numerical values
@@ -237,6 +245,11 @@ public static final <T> Mono<Double> averageDouble(Publisher<T> source, Function
 
 	/**
 	 * Computes the {@link BigInteger} average of items in the source.
+	 * <p>
+	 * The average is computed by summing {@link BigDecimal}-converted values and ultimately
+	 * dividing that number by the number of elements. Eventually the result of that division
+	 * can be rounded or truncated, as the {@link java.math.RoundingMode#FLOOR FLOOR rounding mode}
+	 * is applied and the {@link BigDecimal#toBigInteger()} final conversion is performed.
 	 *
 	 * @param source the numerical source
 	 * @return {@link Mono} of the average of items in source
@@ -248,14 +261,20 @@ public static Mono<BigInteger> averageBigInteger(Publisher<? extends Number> sou
 	/**
 	 * Computes the {@link BigInteger} average of items in the source, which are mapped to
 	 * numerical values using the provided mapping.
+	 * <p>
+	 * Further conversion of individual elements to {@link BigDecimal} is always applied.
+	 * The average is computed by summing {@link BigDecimal}-converted values and ultimately
+	 * dividing that number by the number of elements. Eventually the result of that division
+	 * can be rounded or truncated, as the {@link java.math.RoundingMode#FLOOR FLOOR rounding mode}
+	 * is applied and the {@link BigDecimal#toBigInteger()} final conversion is performed.
 	 *
 	 * @param source  the source items
 	 * @param mapping a function to map source items to numerical values
 	 * @return {@link Mono} of the average of items in source
 	 */
 	public static final <T> Mono<BigInteger> averageBigInteger(Publisher<T> source,
 			Function<? super T, ? extends Number> mapping) {
-		return MathMono.onAssembly(new MonoAverageBigInteger<T>(source, mapping));
+		return MathMono.onAssembly(new MonoAverageBigInteger<>(source, mapping));
 	}
 
 

reactor-extra/src/main/java/reactor/math/MonoAverageBigInteger.java

@@ -16,7 +16,9 @@
 
 package reactor.math;
 
+import java.math.BigDecimal;
 import java.math.BigInteger;
+import java.math.RoundingMode;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
@@ -52,7 +54,7 @@ private static final class AverageBigIntegerSubscriber<T>
 
 		private int count;
 
-		private BigInteger sum = BigInteger.ZERO;
+		private BigDecimal sum = BigDecimal.ZERO;
 
 		AverageBigIntegerSubscriber(CoreSubscriber<? super BigInteger> actual,
 				Function<? super T, ? extends Number> mapping) {
@@ -63,19 +65,28 @@ private static final class AverageBigIntegerSubscriber<T>
 		@Override
 		protected void reset() {
 			count = 0;
-			sum = BigInteger.ZERO;
+			sum = BigDecimal.ZERO;
 		}
 
 		@Override
 		protected BigInteger result() {
-			return (count == 0 ? null : sum.divide(BigInteger.valueOf(count)));
+			return (count == 0 ? null : sum.divide(BigDecimal.valueOf(count), RoundingMode.FLOOR).toBigInteger());
 		}
 
 		@Override
 		protected void updateResult(T newValue) {
 			Number number = mapping.apply(newValue);
-			BigInteger bigIntegerValue = BigInteger.valueOf(number.longValue());
-			sum = sum.add(bigIntegerValue);
+			BigDecimal bigDecimalValue;
+			if (number instanceof BigDecimal) {
+				bigDecimalValue = (BigDecimal) number;
+			}
+			else if (number instanceof BigInteger) {
+				bigDecimalValue = new BigDecimal((BigInteger) number);
+			}
+			else {
+				bigDecimalValue = new BigDecimal(number.toString());
+			}
+			sum = sum.add(bigDecimalValue);
 			count++;
 		}
 	}

reactor-extra/src/main/java/reactor/math/MonoSumBigInteger.java

@@ -16,6 +16,7 @@
 
 package reactor.math;
 
+import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.util.function.Function;
 
@@ -50,7 +51,7 @@ static final private class SumBigIntegerSubscriber<T>
 
 		private final Function<? super T, ? extends Number> mapping;
 
-		BigInteger sum;
+		BigDecimal sum;
 
 		boolean hasValue;
 
@@ -62,20 +63,29 @@ static final private class SumBigIntegerSubscriber<T>
 
 		@Override
 		protected void reset() {
-			sum = BigInteger.ZERO;
+			sum = BigDecimal.ZERO;
 			hasValue = false;
 		}
 
 		@Override
 		protected BigInteger result() {
-			return (hasValue ? sum : null);
+			return (hasValue ? sum.toBigInteger() : null);
 		}
 
 		@Override
 		protected void updateResult(T newValue) {
 			Number number = mapping.apply(newValue);
-			BigInteger bigIntegerValue = BigInteger.valueOf(number.longValue());
-			sum = hasValue ? sum.add(bigIntegerValue) : bigIntegerValue;
+			BigDecimal bigDecimalValue;
+			if (number instanceof BigDecimal) {
+				bigDecimalValue = (BigDecimal) number;
+			}
+			else if (number instanceof BigInteger) {
+				bigDecimalValue = new BigDecimal((BigInteger) number);
+			}
+			else {
+				bigDecimalValue = new BigDecimal(number.toString());
+			}
+			sum = hasValue ? sum.add(bigDecimalValue) : bigDecimalValue;
 			hasValue = true;
 		}
 	}

reactor-extra/src/test/java/reactor/math/ReactorMathTests.java

@@ -177,6 +177,41 @@ public void emptySumBigInteger() {
 		verifyEmptyResult(MathFlux.sumBigInteger(Mono.empty()));
 	}
 
+	@Test
+	public void noOverflowSumBigInteger() {
+		long longValue = 1100000000000000L;
+		String bigValue = "12319800000000000000";
+		verifyResult(MathFlux.sumBigInteger(Mono.just(BigInteger.valueOf(longValue))), BigInteger.valueOf(longValue));
+		verifyResult(MathFlux.sumBigInteger(Mono.just(new BigInteger(bigValue, 10))), new BigInteger(bigValue, 10));
+		verifyResult(
+				MathFlux.sumBigInteger(Flux.just(BigInteger.valueOf(longValue), new BigInteger(bigValue, 10))),
+				new BigInteger(bigValue, 10).add(BigInteger.valueOf(longValue))
+		);
+	}
+
+	@Test
+	public void noProgressivePrecisionLossSumBigInteger() {
+		//smaller than 0.5d, but the fractional parts are individually taken into account
+		double overflowingValue1 = 0.2d;
+		double overflowingValue2 = 0.4d;
+		double overflowingValue3 = 0.4d;
+
+		verifyResult(MathFlux.sumBigInteger(Flux.just(Long.MAX_VALUE, overflowingValue1, overflowingValue2, overflowingValue3)),
+			BigInteger.valueOf(Long.MAX_VALUE).add(BigInteger.ONE));
+	}
+
+	@Test
+	public void finalPrecisionLossSumBigInteger() {
+		//smaller than 0.5d, but the fractional parts are individually taken into account
+		//but at the end the extra 0.2 is dropped
+		double overflowingValue1 = 0.4d;
+		double overflowingValue2 = 0.4d;
+		double overflowingValue3 = 0.4d;
+
+		verifyResult(MathFlux.sumBigInteger(Flux.just(Long.MAX_VALUE, overflowingValue1, overflowingValue2, overflowingValue3)),
+			BigInteger.valueOf(Long.MAX_VALUE).add(BigInteger.ONE));
+	}
+
 	@Test
 	public void fluxSumBigDecimal() {
 		int count = 10;
@@ -317,6 +352,33 @@ public void emptyAverageBigInteger() {
 		verifyEmptyResult(MathFlux.averageBigInteger(Mono.empty()));
 	}
 
+	@Test
+	public void noOverflowAverageBigInteger() {
+		long longValue = 1100000000000000L;
+		String bigValue = "12319800000000000000";
+		verifyResult(MathFlux.averageBigInteger(Mono.just(BigInteger.valueOf(longValue))), BigInteger.valueOf(longValue));
+		verifyResult(MathFlux.averageBigInteger(Mono.just(new BigInteger(bigValue, 10))), new BigInteger(bigValue, 10));
+		verifyResult(
+				MathFlux.averageBigInteger(Flux.just(BigInteger.valueOf(longValue), new BigInteger(bigValue, 10))),
+				new BigInteger(bigValue, 10).add(BigInteger.valueOf(longValue)).divide(BigInteger.valueOf(2L))
+		);
+	}
+
+	@Test
+	public void noProgressivePrecisionLossAverageBigInteger() {
+		Flux<Number> numbers = Flux.just(1001.4d, 1000.2d, 1001.4d);
+		//3003 / 3 == 1001. rounding down each number would instead result in 3002 / 3 == 1000
+
+		verifyResult(MathFlux.averageBigInteger(numbers), BigInteger.valueOf(1001));
+	}
+
+	@Test
+	public void finalPrecisionLossAverageBigInteger() {
+		Flux<Number> numbers = Flux.just(1001.4d, 1000.2d, 1000.4d);
+
+		verifyResult(MathFlux.averageBigInteger(numbers), BigInteger.valueOf(1000));
+	}
+
 	@Test
 	public void fluxAverageBigDecimal() {
 		int count = 10;