Add RedisAtomicDouble

DATAREDIS-198

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

@@ -0,0 +1,321 @@
+/*
+ * Copyright 2013 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.redis.support.atomic;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.data.redis.connection.DataType;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.BoundKeyOperations;
+import org.springframework.data.redis.core.RedisOperations;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.SessionCallback;
+import org.springframework.data.redis.core.ValueOperations;
+import org.springframework.data.redis.serializer.GenericToStringSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.util.Assert;
+
+/**
+ * Atomic double backed by Redis. Uses Redis atomic increment/decrement and watch/multi/exec
+ * operations for CAS operations.
+ *
+ * @author Jennifer Hickey
+ */
+@SuppressWarnings("serial")
+public class RedisAtomicDouble extends Number implements Serializable, BoundKeyOperations<String> {
+
+	private volatile String key;
+	private ValueOperations<String, Double> operations;
+	private RedisOperations<String, Double> generalOps;
+
+	/**
+	 * Constructs a new <code>RedisAtomicDouble</code> instance. Uses the value existing in Redis or
+	 * 0 if none is found.
+	 *
+	 * @param redisCounter
+	 *            redis counter
+	 * @param factory
+	 *            connection factory
+	 */
+	public RedisAtomicDouble(String redisCounter, RedisConnectionFactory factory) {
+		this(redisCounter, factory, null);
+	}
+
+	/**
+	 * Constructs a new <code>RedisAtomicDouble</code> instance.
+	 *
+	 * @param redisCounter
+	 * @param factory
+	 * @param initialValue
+	 */
+	public RedisAtomicDouble(String redisCounter, RedisConnectionFactory factory,
+			double initialValue) {
+		this(redisCounter, factory, Double.valueOf(initialValue));
+	}
+
+	private RedisAtomicDouble(String redisCounter, RedisConnectionFactory factory,
+			Double initialValue) {
+		Assert.hasText(redisCounter, "a valid counter name is required");
+		Assert.notNull(factory, "a valid factory is required");
+
+		RedisTemplate<String, Double> redisTemplate = new RedisTemplate<String, Double>();
+		redisTemplate.setKeySerializer(new StringRedisSerializer());
+		redisTemplate.setValueSerializer(new GenericToStringSerializer<Double>(Double.class));
+		redisTemplate.setExposeConnection(true);
+		redisTemplate.setConnectionFactory(factory);
+		redisTemplate.afterPropertiesSet();
+
+		this.key = redisCounter;
+		this.generalOps = redisTemplate;
+		this.operations = generalOps.opsForValue();
+
+		if (initialValue == null) {
+			if (this.operations.get(redisCounter) == null) {
+				set(0);
+			}
+		} else {
+			set(initialValue);
+		}
+	}
+
+	/**
+	 * Constructs a new <code>RedisAtomicDouble</code> instance. Uses the value existing in Redis or
+	 * 0 if none is found.
+	 *
+	 * @param redisCounter
+	 *            the redis counter
+	 * @param template
+	 *            the template
+	 */
+	public RedisAtomicDouble(String redisCounter, RedisOperations<String, Double> template) {
+		this(redisCounter, template, null);
+	}
+
+	/**
+	 * Constructs a new <code>RedisAtomicDouble</code> instance.
+	 *
+	 * @param redisCounter
+	 *            the redis counter
+	 * @param template
+	 *            the template
+	 * @param initialValue
+	 *            the initial value
+	 */
+	public RedisAtomicDouble(String redisCounter, RedisOperations<String, Double> template,
+			double initialValue) {
+		this(redisCounter, template, Double.valueOf(initialValue));
+	}
+
+	private RedisAtomicDouble(String redisCounter, RedisOperations<String, Double> template,
+			Double initialValue) {
+		Assert.hasText(redisCounter, "a valid counter name is required");
+		Assert.notNull(template, "a valid template is required");
+
+		this.key = redisCounter;
+		this.generalOps = template;
+		this.operations = generalOps.opsForValue();
+
+		if (initialValue == null) {
+			if (this.operations.get(redisCounter) == null) {
+				set(0);
+			}
+		} else {
+			set(initialValue);
+		}
+	}
+
+	/**
+	 * Gets the current value.
+	 *
+	 * @return the current value
+	 */
+	public double get() {
+		return operations.get(key);
+	}
+
+	/**
+	 * Sets to the given value.
+	 *
+	 * @param newValue
+	 *            the new value
+	 */
+	public void set(double newValue) {
+		operations.set(key, newValue);
+	}
+
+	/**
+	 * Atomically sets to the given value and returns the old value.
+	 *
+	 * @param newValue
+	 *            the new value
+	 * @return the previous value
+	 */
+	public double getAndSet(double newValue) {
+		return operations.getAndSet(key, newValue);
+	}
+
+	/**
+	 * Atomically sets the value to the given updated value if the current value {@code ==} the
+	 * expected value.
+	 *
+	 * @param expect
+	 *            the expected value
+	 * @param update
+	 *            the new value
+	 * @return true if successful. False return indicates that the actual value was not equal to the
+	 *         expected value.
+	 */
+	public boolean compareAndSet(final double expect, final double update) {
+		return generalOps.execute(new SessionCallback<Boolean>() {
+
+			@SuppressWarnings({ "unchecked", "rawtypes" })
+			public Boolean execute(RedisOperations operations) {
+				for (;;) {
+					operations.watch(Collections.singleton(key));
+					if (expect == get()) {
+						generalOps.multi();
+						set(update);
+						if (operations.exec() != null) {
+							return true;
+						}
+					}
+					{
+						return false;
+					}
+				}
+			}
+		});
+	}
+
+	/**
+	 * Atomically increments by one the current value.
+	 *
+	 * @return the previous value
+	 */
+	public double getAndIncrement() {
+		return incrementAndGet() - 1.0;
+	}
+
+	/**
+	 * Atomically decrements by one the current value.
+	 *
+	 * @return the previous value
+	 */
+	public double getAndDecrement() {
+		return decrementAndGet() + 1.0;
+	}
+
+	/**
+	 * Atomically adds the given value to the current value.
+	 *
+	 * @param delta
+	 *            the value to add
+	 * @return the previous value
+	 */
+	public double getAndAdd(final double delta) {
+		return addAndGet(delta) - delta;
+	}
+
+	/**
+	 * Atomically increments by one the current value.
+	 *
+	 * @return the updated value
+	 */
+	public double incrementAndGet() {
+		return operations.increment(key, 1.0);
+	}
+
+	/**
+	 * Atomically decrements by one the current value.
+	 *
+	 * @return the updated value
+	 */
+	public double decrementAndGet() {
+		return operations.increment(key, -1.0);
+	}
+
+	/**
+	 * Atomically adds the given value to the current value.
+	 *
+	 * @param delta
+	 *            the value to add
+	 * @return the updated value
+	 */
+	public double addAndGet(double delta) {
+		return operations.increment(key, delta);
+	}
+
+	/**
+	 * Returns the String representation of the current value.
+	 *
+	 * @return the String representation of the current value.
+	 */
+	public String toString() {
+		return Double.toString(get());
+	}
+
+	public String getKey() {
+		return key;
+	}
+
+	public DataType getType() {
+		return DataType.STRING;
+	}
+
+	public Long getExpire() {
+		return generalOps.getExpire(key);
+	}
+
+	public Boolean expire(long timeout, TimeUnit unit) {
+		return generalOps.expire(key, timeout, unit);
+	}
+
+	public Boolean expireAt(Date date) {
+		return generalOps.expireAt(key, date);
+	}
+
+	public Boolean persist() {
+		return generalOps.persist(key);
+	}
+
+	public void rename(String newKey) {
+		generalOps.rename(key, newKey);
+		key = newKey;
+	}
+
+	@Override
+	public double doubleValue() {
+		return get();
+	}
+
+	@Override
+	public float floatValue() {
+		return (float) get();
+	}
+
+	@Override
+	public int intValue() {
+		return (int) get();
+	}
+
+	@Override
+	public long longValue() {
+		return (long) get();
+	}
+}

src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicDoubleTests.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2013 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.redis.support.atomic;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
+
+import java.util.Collection;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.springframework.data.redis.ConnectionFactoryTracker;
+import org.springframework.data.redis.RedisTestProfileValueSource;
+import org.springframework.data.redis.connection.ConnectionUtils;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+
+/**
+ * Integration test of {@link RedisAtomicDouble}
+ *
+ * @author Jennifer Hickey
+ */
+@RunWith(Parameterized.class)
+public class RedisAtomicDoubleTests {
+
+	private RedisAtomicDouble doubleCounter;
+
+	private RedisConnectionFactory factory;
+
+	public RedisAtomicDoubleTests(RedisConnectionFactory factory) {
+		doubleCounter = new RedisAtomicDouble(getClass().getSimpleName() + ":double", factory);
+		this.factory = factory;
+		ConnectionFactoryTracker.add(factory);
+	}
+
+	@Before
+	public void setUp() {
+		// Most atomic Double ops involve incrByFloat, which is new as of 2.6
+		assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "2.6"));
+	}
+
+	@After
+	public void stop() {
+		RedisConnection connection = factory.getConnection();
+		connection.flushDb();
+		connection.close();
+	}
+
+	@AfterClass
+	public static void cleanUp() {
+		ConnectionFactoryTracker.cleanUp();
+	}
+
+	@Parameters
+	public static Collection<Object[]> testParams() {
+		return AtomicCountersParam.testParams();
+	}
+
+	@Test
+	public void testCheckAndSet() {
+		// Txs not supported in Jredis
+		assumeTrue(!ConnectionUtils.isJredis(factory));
+		doubleCounter.set(0);
+		assertFalse(doubleCounter.compareAndSet(1.2, 10.6));
+		assertTrue(doubleCounter.compareAndSet(0, 10.6));
+		assertTrue(doubleCounter.compareAndSet(10.6, 0));
+	}
+
+	@Test
+	public void testIncrementAndGet() throws Exception {
+		assumeTrue(!ConnectionUtils.isJredis(factory) && !(ConnectionUtils.isJedis(factory)));
+		doubleCounter.set(0);
+		assertEquals(1.0, doubleCounter.incrementAndGet(), 0);
+	}
+
+	@Test
+	public void testAddAndGet() throws Exception {
+		assumeTrue(!ConnectionUtils.isJredis(factory) && !(ConnectionUtils.isJedis(factory)));
+		doubleCounter.set(0);
+		double delta = 1.3;
+		assertEquals(delta, doubleCounter.addAndGet(delta), .0001);
+	}
+
+	@Test
+	public void testDecrementAndGet() throws Exception {
+		assumeTrue(!ConnectionUtils.isJredis(factory) && !(ConnectionUtils.isJedis(factory)));
+		doubleCounter.set(1);
+		assertEquals(0, doubleCounter.decrementAndGet(), 0);
+	}
+
+	@Test
+	public void testGetAndSet() {
+		doubleCounter.set(3.4);
+		assertEquals(3.4, doubleCounter.getAndSet(1.2), 0);
+		assertEquals(1.2, doubleCounter.get(), 0);
+	}
+
+	@Test
+	public void testGetAndIncrement() {
+		assumeTrue(!ConnectionUtils.isJredis(factory) && !(ConnectionUtils.isJedis(factory)));
+		doubleCounter.set(2.3);
+		assertEquals(2.3, doubleCounter.getAndIncrement(), 0);
+		assertEquals(3.3, doubleCounter.get(), .0001);
+	}
+
+	@Test
+	public void testGetAndDecrement() {
+		assumeTrue(!ConnectionUtils.isJredis(factory) && !(ConnectionUtils.isJedis(factory)));
+		doubleCounter.set(0.5);
+		assertEquals(0.5, doubleCounter.getAndDecrement(), 0);
+		assertEquals(-0.5, doubleCounter.get(), .0001);
+	}
+
+	@Test
+	public void testGetAndAdd() {
+		assumeTrue(!ConnectionUtils.isJredis(factory) && !(ConnectionUtils.isJedis(factory)));
+		doubleCounter.set(0.5);
+		assertEquals(0.5, doubleCounter.getAndAdd(0.7), 0);
+		assertEquals(1.2, doubleCounter.get(), .0001);
+	}
+
+	@Test
+	public void testExpire() {
+		assertTrue(doubleCounter.expire(1, TimeUnit.SECONDS));
+		assertTrue(doubleCounter.getExpire() > 0);
+	}
+
+	@Test
+	public void testExpireAt() {
+		// JRedis converts Unix time to millis before sending command, so it expires right away
+		assumeTrue(!ConnectionUtils.isJredis(factory));
+		doubleCounter.set(7.8);
+		assertTrue(doubleCounter.expireAt(new Date(System.currentTimeMillis() + 2000)));
+		assertTrue(doubleCounter.getExpire() > 0);
+	}
+
+	@Test
+	public void testRename() {
+		doubleCounter.set(5.6);
+		doubleCounter.rename("foodouble");
+		assertEquals("5.6", new String(factory.getConnection().get("foodouble".getBytes())));
+		assertNull(factory.getConnection().get((getClass().getSimpleName() + ":double").getBytes()));
+	}
+}

src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicIntegerTests.java

@@ -1,12 +1,12 @@
 /*
- * Copyright 2011-2013 the original author or authors.
- * 
+ * Copyright 2013 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.
@@ -37,20 +37,19 @@ import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 
 /**
+ * Integration test of {@link RedisAtomicInteger}
+ *
  * @author Costin Leau
  * @author Jennifer Hickey
  */
 @RunWith(Parameterized.class)
-public class RedisAtomicTests {
+public class RedisAtomicIntegerTests {
 
 	private RedisAtomicInteger intCounter;
-	private RedisAtomicLong longCounter;
 	private RedisConnectionFactory factory;
 
-
-	public RedisAtomicTests(RedisConnectionFactory factory) {
+	public RedisAtomicIntegerTests(RedisConnectionFactory factory) {
 		intCounter = new RedisAtomicInteger(getClass().getSimpleName() + ":int", factory);
-		longCounter = new RedisAtomicLong(getClass().getSimpleName() + ":long", factory);
 		this.factory = factory;
 		ConnectionFactoryTracker.add(factory);
 	}
@@ -73,7 +72,7 @@ public class RedisAtomicTests {
 	}
 
 	@Test
-	public void testIntCheckAndSet() throws Exception {
+	public void testCheckAndSet() throws Exception {
 		// Txs not supported in Jredis
 		assumeTrue(!ConnectionUtils.isJredis(factory));
 		intCounter.set(0);
@@ -83,60 +82,24 @@ public class RedisAtomicTests {
 	}
 
 	@Test
-	public void testLongCheckAndSet() throws Exception {
-		// Txs not supported in Jredis
-		assumeTrue(!ConnectionUtils.isJredis(factory));
-		longCounter.set(0);
-		assertFalse(longCounter.compareAndSet(1, 10));
-		assertTrue(longCounter.compareAndSet(0, 10));
-		assertTrue(longCounter.compareAndSet(10, 0));
-	}
-
-	@Test
-	public void testLongIncrement() throws Exception {
-		longCounter.set(0);
-		assertEquals(1, longCounter.incrementAndGet());
-	}
-
-	@Test
-	public void testIntIncrement() throws Exception {
+	public void testIncrementAndGet() throws Exception {
 		intCounter.set(0);
 		assertEquals(1, intCounter.incrementAndGet());
 	}
 
 	@Test
-	public void testLongCustomIncrement() throws Exception {
-		longCounter.set(0);
-		long delta = 5;
-		assertEquals(delta, longCounter.addAndGet(delta));
-	}
-
-	@Test
-	public void testIntCustomIncrement() throws Exception {
+	public void testAddAndGet() throws Exception {
 		intCounter.set(0);
 		int delta = 5;
 		assertEquals(delta, intCounter.addAndGet(delta));
 	}
 
 	@Test
-	public void testLongDecrement() throws Exception {
-		longCounter.set(1);
-		assertEquals(0, longCounter.decrementAndGet());
-	}
-
-	@Test
-	public void testIntDecrement() throws Exception {
+	public void testDecrementAndGet() throws Exception {
 		intCounter.set(1);
 		assertEquals(0, intCounter.decrementAndGet());
 	}
 
-	@Test
-	public void testReadExistingValue() throws Exception {
-		longCounter.set(5);
-		RedisAtomicLong keyCopy = new RedisAtomicLong(longCounter.getKey(), factory);
-		assertEquals(longCounter.get(), keyCopy.get());
-	}
-
 	@Test
 	@Ignore("DATAREDIS-108 Test is intermittently failing")
 	public void testCompareSet() throws Exception {
@@ -170,4 +133,4 @@ public class RedisAtomicTests {
 
 		assertFalse("counter already modified", failed.get());
 	}
-}
+}

src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicLongTests.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2013 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.redis.support.atomic;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
+
+import java.util.Collection;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.springframework.data.redis.ConnectionFactoryTracker;
+import org.springframework.data.redis.connection.ConnectionUtils;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+
+/**
+ * Integration test of {@link RedisAtomicLong}
+ *
+ * @author Costin Leau
+ * @author Jennifer Hickey
+ */
+@RunWith(Parameterized.class)
+public class RedisAtomicLongTests {
+
+	private RedisAtomicLong longCounter;
+	private RedisConnectionFactory factory;
+
+	public RedisAtomicLongTests(RedisConnectionFactory factory) {
+		longCounter = new RedisAtomicLong(getClass().getSimpleName() + ":long", factory);
+		this.factory = factory;
+		ConnectionFactoryTracker.add(factory);
+	}
+
+	@After
+	public void stop() {
+		RedisConnection connection = factory.getConnection();
+		connection.flushDb();
+		connection.close();
+	}
+
+	@AfterClass
+	public static void cleanUp() {
+		ConnectionFactoryTracker.cleanUp();
+	}
+
+	@Parameters
+	public static Collection<Object[]> testParams() {
+		return AtomicCountersParam.testParams();
+	}
+
+	@Test
+	public void testCheckAndSet() throws Exception {
+		// Txs not supported in Jredis
+		assumeTrue(!ConnectionUtils.isJredis(factory));
+		longCounter.set(0);
+		assertFalse(longCounter.compareAndSet(1, 10));
+		assertTrue(longCounter.compareAndSet(0, 10));
+		assertTrue(longCounter.compareAndSet(10, 0));
+	}
+
+	@Test
+	public void testIncrementAndGet() throws Exception {
+		longCounter.set(0);
+		assertEquals(1, longCounter.incrementAndGet());
+	}
+
+	@Test
+	public void testAddAndGet() throws Exception {
+		longCounter.set(0);
+		long delta = 5;
+		assertEquals(delta, longCounter.addAndGet(delta));
+	}
+
+	@Test
+	public void testDecrementAndGet() throws Exception {
+		longCounter.set(1);
+		assertEquals(0, longCounter.decrementAndGet());
+	}
+
+	@Test
+	public void testGetExistingValue() throws Exception {
+		longCounter.set(5);
+		RedisAtomicLong keyCopy = new RedisAtomicLong(longCounter.getKey(), factory);
+		assertEquals(longCounter.get(), keyCopy.get());
+	}
+}