diff --git a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicDouble.java b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicDouble.java new file mode 100644 index 000000000..b14d20966 --- /dev/null +++ b/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 { + + private volatile String key; + private ValueOperations operations; + private RedisOperations generalOps; + + /** + * Constructs a new RedisAtomicDouble 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 RedisAtomicDouble 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 redisTemplate = new RedisTemplate(); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new GenericToStringSerializer(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 RedisAtomicDouble 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 template) { + this(redisCounter, template, null); + } + + /** + * Constructs a new RedisAtomicDouble instance. + * + * @param redisCounter + * the redis counter + * @param template + * the template + * @param initialValue + * the initial value + */ + public RedisAtomicDouble(String redisCounter, RedisOperations template, + double initialValue) { + this(redisCounter, template, Double.valueOf(initialValue)); + } + + private RedisAtomicDouble(String redisCounter, RedisOperations 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() { + + @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(); + } +} diff --git a/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicDoubleTests.java b/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicDoubleTests.java new file mode 100644 index 000000000..2fb2d51b8 --- /dev/null +++ b/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 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())); + } +} diff --git a/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicTests.java b/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicIntegerTests.java similarity index 71% rename from src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicTests.java rename to src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicIntegerTests.java index 39413890e..a07809dd0 100644 --- a/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicTests.java +++ b/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()); } -} \ No newline at end of file +} diff --git a/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicLongTests.java b/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicLongTests.java new file mode 100644 index 000000000..2335a9881 --- /dev/null +++ b/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 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()); + } +}