DATAREDIS-874 - Implement accumulate/update methods on RedisAtomic[*] classes.

This commit adds the following public methods to RedisAtomicInteger,
RedisAtomicLong, and RedisAtomicDouble:

T accumulateAndGet(T updateValue, BinaryOperator<T> accumulatorFunction);
T getAndAccumulate(T updateValue, BinaryOperator<T> accumulatorFunction);
T updateAndGet(UnaryOperator<T> updateFunction);
T getAndUpdate(UnaryOperator<T> updateFunction);

These methods are primarily useful for doing CAS operations with the
help of Binary/UnaryOperators to generate the new value. They mirror
their counterparts on the AtomicInteger and AtomicLong classes
available in the java.util.concurrent.atomic package.

Original pull request: #362.

src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicDouble.java

@@ -18,6 +18,8 @@ package org.springframework.data.redis.support.atomic;
 import java.io.Serializable;
 import java.util.Date;
 import java.util.concurrent.TimeUnit;
+import java.util.function.DoubleUnaryOperator;
+import java.util.function.DoubleBinaryOperator;
 
 import org.springframework.dao.DataRetrievalFailureException;
 import org.springframework.data.redis.connection.DataType;
@@ -39,6 +41,7 @@ import org.springframework.util.Assert;
  * @author Thomas Darimont
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Graham MacMaster
  */
 public class RedisAtomicDouble extends Number implements Serializable, BoundKeyOperations<String> {
 
@@ -218,6 +221,44 @@ public class RedisAtomicDouble extends Number implements Serializable, BoundKeyO
 		return addAndGet(delta) - delta;
 	}
 
+	/**
+	 * Atomically update the current value using the given update function.
+	 *
+	 * @param updateFunction the function which calculates the value to set. Should be a pure function (no side effects),
+	 *                       because it will be applied several times if update attempts fail due to concurrent calls.
+	 * @return the previous value.
+	 */
+	public double getAndUpdate(DoubleUnaryOperator updateFunction) {
+
+		double previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = updateFunction.applyAsDouble(previousValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return previousValue;
+	}
+
+	/**
+	 * Atomically update the current value using the given accumulator function.
+	 * The new value is calculated by applying the accumulator function to the current value and the
+	 * given `updateValue`.
+	 *
+	 * @param updateValue the value which will be passed into the accumulator function.
+	 * @param accumulatorFunction the function which calculates the value to set. Should be a pure function (no side
+	 *                            effects), because it will be applied several times if update attempts fail due to
+	 *                            concurrent calls.
+	 * @return the previous value.
+	 */
+	public double getAndAccumulate(double updateValue, DoubleBinaryOperator accumulatorFunction) {
+
+		double previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = accumulatorFunction.applyAsDouble(previousValue, updateValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return previousValue;
+	}
+
 	/**
 	 * Atomically increments by one the current value.
 	 *
@@ -246,6 +287,44 @@ public class RedisAtomicDouble extends Number implements Serializable, BoundKeyO
 		return operations.increment(key, delta);
 	}
 
+	/**
+	 * Atomically update the current value using the given update function.
+	 *
+	 * @param updateFunction the function which calculates the value to set. Should be a pure function (no side effects),
+	 *                       because it will be applied several times if update attempts fail due to concurrent calls.
+	 * @return the updated value.
+	 */
+	public double updateAndGet(DoubleUnaryOperator updateFunction) {
+
+		double previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = updateFunction.applyAsDouble(previousValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return newValue;
+	}
+
+	/**
+	 * Atomically update the current value using the given accumulator function.
+	 * The new value is calculated by applying the accumulator function to the current value and the
+	 * given `updateValue`.
+	 *
+	 * @param updateValue the value which will be passed into the accumulator function.
+	 * @param accumulatorFunction the function which calculates the value to set. Should be a pure function (no side
+	 *                            effects), because it will be applied several times if update attempts fail due to
+	 *                            concurrent calls.
+	 * @return the updated value.
+	 */
+	public double accumulateAndGet(double updateValue, DoubleBinaryOperator accumulatorFunction) {
+
+		double previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = accumulatorFunction.applyAsDouble(previousValue, updateValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return newValue;
+	}
+
 	/**
 	 * @return the String representation of the current value.
 	 */

src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicInteger.java

@@ -18,6 +18,8 @@ package org.springframework.data.redis.support.atomic;
 import java.io.Serializable;
 import java.util.Date;
 import java.util.concurrent.TimeUnit;
+import java.util.function.IntBinaryOperator;
+import java.util.function.IntUnaryOperator;
 
 import org.springframework.dao.DataRetrievalFailureException;
 import org.springframework.data.redis.connection.DataType;
@@ -39,6 +41,7 @@ import org.springframework.util.Assert;
  * @author Thomas Darimont
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Graham MacMaster
  * @see java.util.concurrent.atomic.AtomicInteger
  */
 public class RedisAtomicInteger extends Number implements Serializable, BoundKeyOperations<String> {
@@ -216,6 +219,44 @@ public class RedisAtomicInteger extends Number implements Serializable, BoundKey
 		return addAndGet(delta) - delta;
 	}
 
+	/**
+	 * Atomically update the current value using the given update function.
+	 *
+	 * @param updateFunction the function which calculates the value to set. Should be a pure function (no side effects),
+	 *                       because it will be applied several times if update attempts fail due to concurrent calls.
+	 * @return the previous value.
+	 */
+	public int getAndUpdate(IntUnaryOperator updateFunction) {
+
+		int previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = updateFunction.applyAsInt(previousValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return previousValue;
+	}
+
+	/**
+	 * Atomically update the current value using the given accumulator function.
+	 * The new value is calculated by applying the accumulator function to the current value and the
+	 * given `updateValue`.
+	 *
+	 * @param updateValue the value which will be passed into the accumulator function.
+	 * @param accumulatorFunction the function which calculates the value to set. Should be a pure function (no side
+	 *                            effects), because it will be applied several times if update attempts fail due to
+	 *                            concurrent calls.
+	 * @return the previous value.
+	 */
+	public int getAndAccumulate(int updateValue, IntBinaryOperator accumulatorFunction) {
+
+		int previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = accumulatorFunction.applyAsInt(previousValue, updateValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return previousValue;
+	}
+
 	/**
 	 * Atomically increment by one the current value.
 	 *
@@ -244,6 +285,44 @@ public class RedisAtomicInteger extends Number implements Serializable, BoundKey
 		return operations.increment(key, delta).intValue();
 	}
 
+	/**
+	 * Atomically update the current value using the given update function.
+	 *
+	 * @param updateFunction the function which calculates the value to set. Should be a pure function (no side effects),
+	 *                       because it will be applied several times if update attempts fail due to concurrent calls.
+	 * @return the updated value.
+	 */
+	public int updateAndGet(IntUnaryOperator updateFunction) {
+
+		int previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = updateFunction.applyAsInt(previousValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return newValue;
+	}
+
+	/**
+	 * Atomically update the current value using the given accumulator function.
+	 * The new value is calculated by applying the accumulator function to the current value and the
+	 * given `updateValue`.
+	 *
+	 * @param updateValue the value which will be passed into the accumulator function.
+	 * @param accumulatorFunction the function which calculates the value to set. Should be a pure function (no side
+	 *                            effects), because it will be applied several times if update attempts fail due to
+	 *                            concurrent calls.
+	 * @return the updated value.
+	 */
+	public int accumulateAndGet(int updateValue, IntBinaryOperator accumulatorFunction) {
+
+		int previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = accumulatorFunction.applyAsInt(previousValue, updateValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return newValue;
+	}
+
 	/**
 	 * @return the String representation of the current value.
 	 */

src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicLong.java

@@ -18,6 +18,8 @@ package org.springframework.data.redis.support.atomic;
 import java.io.Serializable;
 import java.util.Date;
 import java.util.concurrent.TimeUnit;
+import java.util.function.LongBinaryOperator;
+import java.util.function.LongUnaryOperator;
 
 import org.springframework.dao.DataRetrievalFailureException;
 import org.springframework.data.redis.connection.DataType;
@@ -40,6 +42,7 @@ import org.springframework.util.Assert;
  * @author Thomas Darimont
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Graham MacMaster
  * @see java.util.concurrent.atomic.AtomicLong
  */
 public class RedisAtomicLong extends Number implements Serializable, BoundKeyOperations<String> {
@@ -223,6 +226,44 @@ public class RedisAtomicLong extends Number implements Serializable, BoundKeyOpe
 		return addAndGet(delta) - delta;
 	}
 
+	/**
+	 * Atomically update the current value using the given update function.
+	 *
+	 * @param updateFunction the function which calculates the value to set. Should be a pure function (no side effects),
+	 *                       because it will be applied several times if update attempts fail due to concurrent calls.
+	 * @return the previous value.
+	 */
+	public long getAndUpdate(LongUnaryOperator updateFunction) {
+
+		long previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = updateFunction.applyAsLong(previousValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return previousValue;
+	}
+
+	/**
+	 * Atomically update the current value using the given accumulator function.
+	 * The new value is calculated by applying the accumulator function to the current value and the
+	 * given `updateValue`.
+	 *
+	 * @param updateValue the value which will be passed into the accumulator function.
+	 * @param accumulatorFunction the function which calculates the value to set. Should be a pure function (no side
+	 *                            effects), because it will be applied several times if update attempts fail due to
+	 *                            concurrent calls.
+	 * @return the previous value.
+	 */
+	public long getAndAccumulate(long updateValue, LongBinaryOperator accumulatorFunction) {
+
+		long previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = accumulatorFunction.applyAsLong(previousValue, updateValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return previousValue;
+	}
+
 	/**
 	 * Atomically increments by one the current value.
 	 *
@@ -251,6 +292,44 @@ public class RedisAtomicLong extends Number implements Serializable, BoundKeyOpe
 		return operations.increment(key, delta);
 	}
 
+	/**
+	 * Atomically update the current value using the given update function.
+	 *
+	 * @param updateFunction the function which calculates the value to set. Should be a pure function (no side effects),
+	 *                       because it will be applied several times if update attempts fail due to concurrent calls.
+	 * @return the updated value.
+	 */
+	public long updateAndGet(LongUnaryOperator updateFunction) {
+
+		long previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = updateFunction.applyAsLong(previousValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return newValue;
+	}
+
+	/**
+	 * Atomically update the current value using the given accumulator function.
+	 * The new value is calculated by applying the accumulator function to the current value and the
+	 * given `updateValue`.
+	 *
+	 * @param updateValue the value which will be passed into the accumulator function.
+	 * @param accumulatorFunction the function which calculates the value to set. Should be a pure function (no side
+	 *                            effects), because it will be applied several times if update attempts fail due to
+	 *                            concurrent calls.
+	 * @return the updated value.
+	 */
+	public long accumulateAndGet(long updateValue, LongBinaryOperator accumulatorFunction) {
+
+		long previousValue, newValue;
+		do {
+			previousValue = get();
+			newValue = accumulatorFunction.applyAsLong(previousValue, updateValue);
+		} while (!compareAndSet(previousValue, newValue));
+		return newValue;
+	}
+
 	/**
 	 * @return the String representation of the current value.
 	 */