diff --git a/src/main/java/org/springframework/data/redis/support/atomic/CompareAndSet.java b/src/main/java/org/springframework/data/redis/support/atomic/CompareAndSet.java new file mode 100644 index 000000000..0708d09d0 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/support/atomic/CompareAndSet.java @@ -0,0 +1,80 @@ +/* + * Copyright 2018 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 lombok.RequiredArgsConstructor; + +import java.util.Collection; +import java.util.function.Consumer; +import java.util.function.Supplier; + +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.core.RedisOperations; +import org.springframework.data.redis.core.SessionCallback; +import org.springframework.util.CollectionUtils; + +/** + * Compare-and-set (CAS) operation using Redis Transactions ({@literal WATCH} and {@literal MULTI}) to atomically update + * the value at {@code key}. + *

+ * The CAS block registers a {@literal WATCH} on the key holding the expected value which guarantees that changes after + * watching and comparing the key will rollback the transaction. The {@literal WATCH} is reset if the comparison fails. + * + * @author Mark Paluch + * @since 2.0.8 + * @see RedisAtomicDouble + * @see RedisAtomicInteger + * @see RedisAtomicLong + */ +@RequiredArgsConstructor +class CompareAndSet implements SessionCallback { + + private final Supplier getter; + private final Consumer setter; + private final String key; + private final T expect; + private final T update; + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.SessionCallback#execute(org.springframework.data.redis.core.RedisOperations) + */ + @Override + @SuppressWarnings("unchecked") + public Boolean execute(RedisOperations operations) throws DataAccessException { + + RedisOperations ops = (RedisOperations) operations; + + ops.watch(key); + + if (expect.equals(getter.get())) { + + ops.multi(); + setter.accept(update); + + if (updateSuccessful(operations.exec())) { + return true; + } + } + + ops.unwatch(); + return false; + } + + private static boolean updateSuccessful(Collection exec) { + return !CollectionUtils.isEmpty(exec); + } +} 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 index 4ef78ce46..4ebda552b 100644 --- a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicDouble.java +++ b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicDouble.java @@ -16,7 +16,6 @@ package org.springframework.data.redis.support.atomic; import java.io.Serializable; -import java.util.Collections; import java.util.Date; import java.util.concurrent.TimeUnit; @@ -26,7 +25,6 @@ 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.RedisSerializer; @@ -190,27 +188,7 @@ public class RedisAtomicDouble extends Number implements Serializable, BoundKeyO * expected value. */ public boolean compareAndSet(final double expect, final double update) { - - return generalOps.execute(new SessionCallback() { - - @Override - @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; - } - } - } - }); + return generalOps.execute(new CompareAndSet<>(this::get, this::set, key, expect, update)); } /** diff --git a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicInteger.java b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicInteger.java index f15097661..fd8edfced 100644 --- a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicInteger.java +++ b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicInteger.java @@ -16,7 +16,6 @@ package org.springframework.data.redis.support.atomic; import java.io.Serializable; -import java.util.Collections; import java.util.Date; import java.util.concurrent.TimeUnit; @@ -26,7 +25,6 @@ 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.RedisSerializer; @@ -188,27 +186,7 @@ public class RedisAtomicInteger extends Number implements Serializable, BoundKey * expected value. */ public boolean compareAndSet(int expect, int update) { - - return generalOps.execute(new SessionCallback() { - - @Override - @SuppressWarnings("unchecked") - 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; - } - } - } - }); + return generalOps.execute(new CompareAndSet<>(this::get, this::set, key, expect, update)); } /** diff --git a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicLong.java b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicLong.java index b8a3daf0a..3495e4c69 100644 --- a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicLong.java +++ b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicLong.java @@ -16,7 +16,6 @@ package org.springframework.data.redis.support.atomic; import java.io.Serializable; -import java.util.Collections; import java.util.Date; import java.util.concurrent.TimeUnit; @@ -26,7 +25,6 @@ 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.RedisSerializer; @@ -194,27 +192,7 @@ public class RedisAtomicLong extends Number implements Serializable, BoundKeyOpe * expected value. */ public boolean compareAndSet(long expect, long update) { - - return generalOps.execute(new SessionCallback() { - - @Override - @SuppressWarnings("unchecked") - 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; - } - } - } - }); + return generalOps.execute(new CompareAndSet<>(this::get, this::set, key, expect, update)); } /** diff --git a/src/test/java/org/springframework/data/redis/support/atomic/CompareAndSetIntegrationTests.java b/src/test/java/org/springframework/data/redis/support/atomic/CompareAndSetIntegrationTests.java new file mode 100644 index 000000000..e0e72f639 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/support/atomic/CompareAndSetIntegrationTests.java @@ -0,0 +1,136 @@ +/* + * Copyright 2018 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.assertj.core.api.Assertions.*; + +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.RedisConnection; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.ValueOperations; +import org.springframework.data.redis.serializer.GenericToStringSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; + +/** + * Integration tests for {@link CompareAndSet}. + * + * @author Mark Paluch + */ +@RunWith(Parameterized.class) +public class CompareAndSetIntegrationTests { + + private static final String KEY = "key"; + + private final RedisConnectionFactory factory; + private final RedisTemplate template; + private final ValueOperations valueOps; + + public CompareAndSetIntegrationTests(RedisConnectionFactory factory) { + + this.factory = factory; + + this.template = new RedisTemplate<>(); + this.template.setConnectionFactory(factory); + this.template.setKeySerializer(new StringRedisSerializer()); + this.template.setValueSerializer(new GenericToStringSerializer<>(Long.class)); + this.template.afterPropertiesSet(); + + this.valueOps = this.template.opsForValue(); + + 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 // DATAREDIS-843 + public void shouldUpdateCounter() { + + long expected = 5; + long actual = 5; + long update = 6; + + CompareAndSet cas = new CompareAndSet<>(() -> actual, newValue -> valueOps.set(KEY, newValue), KEY, expected, + update); + + Boolean executed = this.template.execute(cas); + + assertThat(executed).isTrue(); + assertThat(valueOps.get(KEY)).isEqualTo(update); + } + + @Test // DATAREDIS-843 + public void expectationNotMet() { + + long expected = 5; + long actual = 7; + long update = 6; + + CompareAndSet cas = new CompareAndSet<>(() -> actual, newValue -> valueOps.set(KEY, newValue), KEY, expected, + update); + + Boolean executed = this.template.execute(cas); + + assertThat(executed).isFalse(); + assertThat(valueOps.get(KEY)).isNull(); + } + + @Test // DATAREDIS-843 + public void concurrentUpdate() { + + long expected = 5; + long actual = 5; + long update = 6; + long concurrentlyUpdated = 7; + + CompareAndSet cas = new CompareAndSet<>(() -> actual, newValue -> { + + RedisConnection connection = this.factory.getConnection(); + connection.set(KEY.getBytes(), Long.toString(concurrentlyUpdated).getBytes()); + connection.close(); + + valueOps.set(KEY, newValue); + }, KEY, expected, update); + + Boolean executed = this.template.execute(cas); + + assertThat(executed).isFalse(); + assertThat(valueOps.get(KEY)).isEqualTo(concurrentlyUpdated); + } +} 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 index 44e38a1d7..8219a2eda 100644 --- a/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicDoubleTests.java +++ b/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicDoubleTests.java @@ -69,12 +69,6 @@ public class RedisAtomicDoubleTests extends AbstractRedisAtomicsTests { 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();