diff --git a/.gitignore b/.gitignore
index 4b939dc033..11fee3f1a0 100644
--- a/.gitignore
+++ b/.gitignore
@@ -28,3 +28,4 @@ target
vf.gf.dmn-*
/atlassian-ide-plugin.xml
hostkey.ser
+.springBeans
diff --git a/build.gradle b/build.gradle
index 98e4fc3dd7..e1df0e6656 100644
--- a/build.gradle
+++ b/build.gradle
@@ -96,6 +96,7 @@ subprojects { subproject ->
saajApiVersion = '1.3.5'
saajImplVersion = '1.3.23'
servletApiVersion = '3.1.0'
+ slf4jVersion = "1.7.6"
smackVersion = '3.2.1'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.3.1.RELEASE'
springDataMongoVersion = '1.1.1.RELEASE'
@@ -453,6 +454,7 @@ project('spring-integration-redis') {
exclude group: 'org.springframework'
}
testCompile "com.lambdaworks:lettuce:$lettuceVersion"
+ testCompile "org.slf4j:slf4j-log4j12:$slf4jVersion"
}
}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
new file mode 100644
index 0000000000..40215f3f57
--- /dev/null
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
@@ -0,0 +1,485 @@
+/*
+ * Copyright 2014 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.integration.redis.util;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.serializer.SerializationException;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.integration.util.DefaultLockRegistry;
+import org.springframework.integration.util.LockRegistry;
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@link LockRegistry} providing a distributed lock using Redis.
+ * Locks are stored under the key {@code registryKey:lockKey}. Locks expire after
+ * (default 60) seconds. Threads unlocking an
+ * expired lock will get an {@link IllegalStateException}. This should be
+ * considered as a critical error because it is possible the protected
+ * resources were compromised.
+ *
+ * Locks are reentrant.
+ *
+ * However, locks are scoped by the registry; a lock from a different registry with the
+ * same key (even if the registry uses the same 'registryKey') are different
+ * locks, and the second cannot be acquired by the same thread while the first is
+ * locked.
+ *
+ * Note: This is not intended for low latency applications. It is intended
+ * for resource locking across multiple JVMs.
+ * When a lock is released by a remote system, waiting threads may take up to 100ms
+ * to acquire the lock.
+ * A more performant version would need to get notifications from the Redis stores
+ * of key changes. This is currently only available using the SYNC command.
+ *
+ * This limitation will usually not apply when a lock is released within this registry,
+ * unless another system takes the lock after the local lock is acquired here.
+ * A {@link DefaultLockRegistry} is used internally to achieve this optimization.
+ *
+ * {@link Condition}s are not supported.
+ *
+ * @author Gary Russell
+ * @since 4.0
+ *
+ */
+public final class RedisLockRegistry implements LockRegistry {
+
+ private static final Log logger = LogFactory.getLog(LockRegistry.class);
+
+ private static final byte[] hostName;
+
+ private static final long DEFAULT_EXPIRE_AFTER = 60000;
+
+ private final String registryKey;
+
+ private final RedisTemplate redisTemplate;
+
+ private final ThreadLocal> threadLocks = new ThreadLocal>();
+
+ private final long expireAfter;
+
+ private final LockRegistry localRegistry = new DefaultLockRegistry();
+
+ static {
+ String host;
+ try {
+ host = InetAddress.getLocalHost().getHostName();
+ }
+ catch (UnknownHostException e) {
+ host = "unknownHost";
+ }
+ hostName = host.getBytes();
+ }
+
+ /**
+ * Constructs a lock registry with the default (60 second) lock expiration.
+ * @param connectionFactory The connection factory.
+ * @param registryKey The key prefix for locks.
+ */
+ public RedisLockRegistry(RedisConnectionFactory connectionFactory, String registryKey) {
+ this(connectionFactory, registryKey, DEFAULT_EXPIRE_AFTER);
+ }
+
+ /**
+ * Constructs a lock registry with the supplied lock expiration.
+ * @param connectionFactory The connection factory.
+ * @param registryKey The key prefix for locks.
+ * @param expireAfter The expiration in milliseconds.
+ */
+ public RedisLockRegistry(RedisConnectionFactory connectionFactory, String registryKey, long expireAfter) {
+ Assert.notNull(connectionFactory, "'connectionFactory' cannot be null");
+ Assert.notNull(registryKey, "'registryKey' cannot be null");
+ this.redisTemplate = new RedisTemplate();
+ this.redisTemplate.setConnectionFactory(connectionFactory);
+ this.redisTemplate.setKeySerializer(new StringRedisSerializer());
+ this.redisTemplate.setValueSerializer(new LockSerializer());
+ this.redisTemplate.afterPropertiesSet();
+ this.registryKey = registryKey;
+ this.expireAfter = expireAfter;
+ }
+
+ @Override
+ public Lock obtain(Object lockKey) {
+ Assert.isInstanceOf(String.class, lockKey);
+ List locks = this.threadLocks.get();
+ if (locks == null) {
+ locks = new LinkedList();
+ this.threadLocks.set(locks);
+ }
+ RedisLock lock = null;
+ for (RedisLock alock : locks) {
+ if (alock.getLockKey().equals(lockKey)) {
+ lock = alock;
+ break;
+ }
+ }
+ /*
+ * If the lock is locked, check that it matches what's in the store.
+ * If it doesn't, the lock must have expired.
+ */
+ if (lock != null && lock.thread != null) {
+ RedisLock lockInStore = RedisLockRegistry.this.redisTemplate
+ .boundValueOps(this.registryKey + ":" + lockKey).get();
+ if (lockInStore == null || !lock.equals(lockInStore)) {
+ removeLockFromThreadLocal(locks, lock);
+ lock = null;
+ }
+ }
+ if (lock == null) {
+ lock = new RedisLock((String) lockKey);
+ locks.add(lock);
+ }
+ return lock;
+ }
+
+ private void removeLockFromThreadLocal(List locks, RedisLock lock) {
+ Iterator iterator = locks.iterator();
+ while (iterator.hasNext()) {
+ if (iterator.next().equals(lock)) {
+ iterator.remove();
+ break;
+ }
+ }
+ }
+
+ public Collection listLocks() {
+ Set keys = this.redisTemplate.keys(this.registryKey + ":*");
+ List locks = new ArrayList(keys.size());
+ for (String key : keys) {
+ RedisLock lock = this.redisTemplate.boundValueOps(key).get();
+ if (lock != null) {
+ locks.add(lock);
+ }
+ }
+ return locks;
+ }
+
+ private class RedisLock implements Lock {
+
+ private final String lockKey;
+
+ private long lockedAt;
+
+ private Thread thread;
+
+ private String threadName;
+
+ private byte[] lockHost;
+
+ private int reLock;
+
+ private RedisLock(String lockKey) {
+ this.lockKey = lockKey;
+ this.lockHost = RedisLockRegistry.hostName;
+ }
+
+ private String getLockKey() {
+ return lockKey;
+ }
+
+ @Override
+ public void lock() {
+ Lock localLock = RedisLockRegistry.this.localRegistry.obtain(lockKey);
+ localLock.lock();
+ try {
+ while (true) {
+ try {
+ while (!this.obtainLock()) {
+ Thread.sleep(100);
+ }
+ break;
+ }
+ catch (InterruptedException e) {
+ /*
+ * This method must be uninterruptible so catch and ignore
+ * interrupts and only break out of the while loop when
+ * we get the lock.
+ */
+ }
+ }
+ }
+ catch (Exception e) {
+ localLock.unlock();
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void lockInterruptibly() throws InterruptedException {
+ Lock localLock = RedisLockRegistry.this.localRegistry.obtain(lockKey);
+ localLock.lockInterruptibly();
+ try {
+ while (!this.obtainLock()) {
+ Thread.sleep(100);
+ }
+ }
+ catch (InterruptedException ie) {
+ localLock.unlock();
+ throw ie;
+ }
+ catch (Exception e) {
+ localLock.unlock();
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public boolean tryLock() {
+ Lock localLock = RedisLockRegistry.this.localRegistry.obtain(lockKey);
+ try {
+ if (!localLock.tryLock()) {
+ return false;
+ }
+ boolean obtainedLock = this.obtainLock();
+ if (!obtainedLock) {
+ localLock.unlock();
+ }
+ return obtainedLock;
+ }
+ catch (Exception e) {
+ localLock.unlock();
+ throw new RuntimeException(e);
+ }
+ }
+
+ private boolean obtainLock() {
+ Thread currentThread = Thread.currentThread();
+ if (currentThread.equals(this.thread)) {
+ this.reLock++;
+ return true;
+ }
+ /*
+ * Set these now so they will be persisted if successful.
+ */
+ this.lockedAt = System.currentTimeMillis();
+ this.threadName = currentThread.getName();
+ Boolean success = RedisLockRegistry.this.redisTemplate.boundValueOps(
+ constructLockKey()).setIfAbsent(this);
+ if (!success) {
+ this.lockedAt = 0;
+ this.threadName = null;
+ }
+ else {
+ this.thread = currentThread;
+ RedisLockRegistry.this.redisTemplate.expire(constructLockKey(),
+ RedisLockRegistry.this.expireAfter, TimeUnit.MILLISECONDS);
+ if (logger.isDebugEnabled()) {
+ logger.debug("New lock; " + this.toString());
+ }
+ }
+ return success;
+ }
+
+ @Override
+ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
+ Lock localLock = RedisLockRegistry.this.localRegistry.obtain(lockKey);
+ if (!localLock.tryLock(time, unit)) {
+ return false;
+ }
+ try {
+ long expire = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(time, unit);
+ boolean acquired = false;
+ while (!(acquired = this.obtainLock()) && System.currentTimeMillis() < expire) {
+ Thread.sleep(100);
+ }
+ if (!acquired) {
+ localLock.unlock();
+ }
+ return acquired;
+ }
+ catch (Exception e) {
+ localLock.unlock();
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void unlock() {
+ if (!Thread.currentThread().equals(this.thread)) {
+ if (this.thread == null) {
+ throw new IllegalStateException("Lock is not locked; " + this.toString());
+ }
+ throw new IllegalStateException("Lock is owned by " + this.thread.getName() + "; " + this.toString());
+ }
+ try {
+ if (this.reLock-- <= 0) {
+ List locks = RedisLockRegistry.this.threadLocks.get();
+ if (locks != null) {
+ removeLockFromThreadLocal(locks, this);
+ if (locks.size() == 0) { // last lock for this thread
+ RedisLockRegistry.this.threadLocks.remove();
+ }
+ }
+ this.assertLockInRedisIsUnchanged();
+ RedisLockRegistry.this.redisTemplate.delete(constructLockKey());
+ if (logger.isDebugEnabled()) {
+ logger.debug("Released lock; " + this.toString());
+ }
+ this.thread = null;
+ this.reLock = 0;
+ }
+ }
+ finally {
+ Lock localLock = RedisLockRegistry.this.localRegistry.obtain(lockKey);
+ localLock.unlock();
+ }
+ }
+
+ private void assertLockInRedisIsUnchanged() {
+ RedisLock lockInStore = RedisLockRegistry.this.redisTemplate.boundValueOps(
+ constructLockKey()).get();
+ if (lockInStore == null || !this.equals(lockInStore)) {
+ throw new IllegalStateException("Lock was released due to expiration; " + this.toString()
+ + (lockInStore == null ? "" : "; lock in store: " + lockInStore.toString()));
+ }
+ }
+
+ private String constructLockKey() {
+ return RedisLockRegistry.this.registryKey + ":" + this.lockKey;
+ }
+
+ @Override
+ public Condition newCondition() {
+ throw new UnsupportedOperationException("Conditions are not supported");
+ }
+
+ @Override
+ public String toString() {
+ SimpleDateFormat dateFormat = new SimpleDateFormat("YYYY-MM-dd@HH:mm:ss.SSS");
+ return "RedisLock [lockKey=" + constructLockKey()
+ + ",lockedAt=" + dateFormat.format(new Date(this.lockedAt))
+ + ", thread=" + this.threadName
+ + ", lockHost=" + new String(this.lockHost)
+ + "]";
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + getOuterType().hashCode();
+ result = prime * result + Arrays.hashCode(lockHost);
+ result = prime * result + ((lockKey == null) ? 0 : lockKey.hashCode());
+ result = prime * result + (int) (lockedAt ^ (lockedAt >>> 32));
+ result = prime * result + ((threadName == null) ? 0 : threadName.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null) {
+ return false;
+ }
+ if (getClass() != obj.getClass()) {
+ return false;
+ }
+ RedisLock other = (RedisLock) obj;
+ if (!getOuterType().equals(other.getOuterType())) {
+ return false;
+ }
+ if (!Arrays.equals(lockHost, other.lockHost)) {
+ return false;
+ }
+ if (!lockKey.equals(other.lockKey)) {
+ return false;
+ }
+ if (lockedAt != other.lockedAt) {
+ return false;
+ }
+ if (threadName == null) {
+ if (other.threadName != null) {
+ return false;
+ }
+ }
+ else if (!threadName.equals(other.threadName)) {
+ return false;
+ }
+ return true;
+ }
+
+ private RedisLockRegistry getOuterType() {
+ return RedisLockRegistry.this;
+ }
+
+ }
+
+ private class LockSerializer implements RedisSerializer {
+
+ @Override
+ public byte[] serialize(RedisLock t) throws SerializationException {
+ int hostLength = t.lockHost.length;
+ int keyLength = t.lockKey.length();
+ int threadNameLength = t.threadName.length();
+ byte[] value = new byte[1 + hostLength +
+ 1 + keyLength +
+ 1 + threadNameLength + 8];
+ ByteBuffer buff = ByteBuffer.wrap(value);
+ buff.put((byte) hostLength)
+ .put(t.lockHost)
+ .put((byte) keyLength)
+ .put(t.lockKey.getBytes())
+ .put((byte) threadNameLength)
+ .put(t.threadName.getBytes())
+ .putLong(t.lockedAt);
+ return value;
+ }
+
+ @Override
+ public RedisLock deserialize(byte[] bytes) throws SerializationException {
+ if (bytes == null) {
+ return null;
+ }
+ ByteBuffer buff = ByteBuffer.wrap(bytes);
+ byte[] host = new byte[buff.get()];
+ buff.get(host);
+ byte[] lockKey = new byte[buff.get()];
+ buff.get(lockKey);
+ byte[] threadName = new byte[buff.get()];
+ buff.get(threadName);
+ long lockedAt = buff.getLong();
+ RedisLock lock = new RedisLock(new String(lockKey));
+ lock.lockedAt = lockedAt;
+ lock.lockHost = host;
+ lock.threadName = new String(threadName);
+ return lock;
+ }
+
+ }
+
+}
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/package-info.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/package-info.java
new file mode 100644
index 0000000000..768b316bd2
--- /dev/null
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/package-info.java
@@ -0,0 +1,4 @@
+/**
+ * Provides utility classes.
+ */
+package org.springframework.integration.redis.util;
diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java
index 96d301d9b2..6a82c43e63 100644
--- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java
+++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -44,7 +44,7 @@ public class RedisAvailableTests {
@Rule
public RedisAvailableRule redisAvailableRule = new RedisAvailableRule();
- protected RedisConnectionFactory getConnectionFactoryForTest(){
+ protected RedisConnectionFactory getConnectionFactoryForTest() {
LettuceConnectionFactory connectionFactory = RedisAvailableRule.connectionFactoryResource.get();
RedisTemplate rt = new RedisTemplate();
rt.setConnectionFactory(connectionFactory);
diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests-context.xml
new file mode 100644
index 0000000000..4b141100ab
--- /dev/null
+++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests-context.xml
@@ -0,0 +1,42 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java
new file mode 100644
index 0000000000..63c8216faf
--- /dev/null
+++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java
@@ -0,0 +1,226 @@
+/*
+ * Copyright 2014 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.integration.redis.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.aggregator.ReleaseStrategy;
+import org.springframework.integration.redis.rules.RedisAvailable;
+import org.springframework.integration.redis.rules.RedisAvailableTests;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ * @since 4.0
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class AggregatorWithRedisLocksTests extends RedisAvailableTests {
+
+ @Autowired
+ private LatchingReleaseStrategy releaseStrategy;
+
+ @Autowired
+ private MessageChannel in;
+
+ @Autowired
+ private MessageChannel in2;
+
+ @Autowired
+ private PollableChannel out;
+
+ private volatile Exception exception;
+
+ private RedisTemplate template;
+
+ @Before
+ @After
+ public void setup() {
+ this.template = this.createTemplate();
+ Set keys = template.keys("aggregatorWithRedisLocksTests:*");
+ for (String key : keys) {
+ template.delete(key);
+ }
+ }
+
+ @Test
+ @RedisAvailable
+ public void testLockSingleGroup() throws Exception {
+ this.releaseStrategy.reset(1);
+ Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1));
+ Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 1));
+ assertTrue(this.releaseStrategy.latch2.await(10, TimeUnit.SECONDS));
+ assertEquals(1, this.template.keys("aggregatorWithRedisLocksTests:*").size());
+ this.releaseStrategy.latch1.countDown();
+ assertNotNull(this.out.receive(10000));
+ assertEquals(1, this.releaseStrategy.maxCallers.get());
+ this.assertNoLocksAfterTest();
+ assertNull("Unexpected exception:" + (this.exception != null ? this.exception.toString() : ""), this.exception);
+ }
+
+ @Test
+ @RedisAvailable
+ public void testLockThreeGroups() throws Exception {
+ this.releaseStrategy.reset(3);
+ Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1));
+ Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 1));
+ Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 2));
+ Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 2));
+ Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 3));
+ Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 3));
+ assertTrue(this.releaseStrategy.latch2.await(10, TimeUnit.SECONDS));
+ assertEquals(3, this.template.keys("aggregatorWithRedisLocksTests:*").size());
+ this.releaseStrategy.latch1.countDown();
+ this.releaseStrategy.latch1.countDown();
+ this.releaseStrategy.latch1.countDown();
+ assertNotNull(this.out.receive(10000));
+ assertNotNull(this.out.receive(10000));
+ assertNotNull(this.out.receive(10000));
+ assertEquals(3, this.releaseStrategy.maxCallers.get());
+ this.assertNoLocksAfterTest();
+ assertNull("Unexpected exception:" + (this.exception != null ? this.exception.toString() : ""), this.exception);
+ }
+
+ @Test
+ @RedisAvailable
+ public void testDistributedAggregator() throws Exception {
+ this.releaseStrategy.reset(1);
+ Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1));
+ Executors.newSingleThreadExecutor().execute(new Runnable() {
+
+ @Override
+ public void run() {
+ try {
+ in2.send(new GenericMessage("bar", stubHeaders(2, 2, 1)));
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ exception = e;
+ }
+ }
+ });
+ assertTrue(this.releaseStrategy.latch2.await(10, TimeUnit.SECONDS));
+ assertEquals(1, this.template.keys("aggregatorWithRedisLocksTests:*").size());
+ this.releaseStrategy.latch1.countDown();
+ assertNotNull(this.out.receive(10000));
+ assertEquals(1, this.releaseStrategy.maxCallers.get());
+ this.assertNoLocksAfterTest();
+ assertNull("Unexpected exception:" + (this.exception != null ? this.exception.toString() : ""), this.exception);
+ }
+
+ private void assertNoLocksAfterTest() throws Exception {
+ int n = 0;
+ while (n++ < 100 && this.template.keys("aggregatorWithRedisLocksTests:*").size() > 0) {
+ Thread.sleep(100);
+ }
+ assertEquals(0, this.template.keys("aggregatorWithRedisLocksTests:*").size());
+ }
+
+ private Runnable asyncSend(final String payload, final int sequence, final int correlation) {
+ return new Runnable() {
+
+ @Override
+ public void run() {
+ try {
+ in.send(new GenericMessage(payload, stubHeaders(sequence, 2, correlation)));
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ exception = e;
+ }
+ }
+ };
+ }
+
+ private Map stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) {
+ Map headers = new HashMap();
+ headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber);
+ headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize);
+ headers.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId);
+ return headers;
+ }
+
+ private RedisTemplate createTemplate() {
+ RedisTemplate template = new RedisTemplate();
+ template.setConnectionFactory(this.getConnectionFactoryForTest());
+ template.setKeySerializer(new StringRedisSerializer());
+ template.afterPropertiesSet();
+ return template;
+ }
+
+ public static class LatchingReleaseStrategy implements ReleaseStrategy {
+
+ private volatile CountDownLatch latch1;
+
+ private volatile CountDownLatch latch2;
+
+ private volatile AtomicInteger callers;
+
+ private volatile AtomicInteger maxCallers;
+
+ @Override
+ public boolean canRelease(MessageGroup group) {
+ synchronized(this) {
+ this.callers.incrementAndGet();
+ this.maxCallers.set(Math.max(this.maxCallers.get(), this.callers.get()));
+ }
+ this.latch2.countDown();
+ try {
+ this.latch1.await(10, TimeUnit.SECONDS);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ this.callers.decrementAndGet();
+ return group.size() > 1;
+ }
+
+ public void reset(int expectedConcurrency) {
+ this.latch1 = new CountDownLatch(expectedConcurrency);
+ this.latch2 = new CountDownLatch(expectedConcurrency);
+ this.callers = new AtomicInteger();
+ this.maxCallers = new AtomicInteger();
+ }
+ }
+}
diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java
new file mode 100644
index 0000000000..6868accd87
--- /dev/null
+++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java
@@ -0,0 +1,431 @@
+/*
+ * Copyright 2014 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.integration.redis.util;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Collection;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.integration.redis.rules.RedisAvailable;
+import org.springframework.integration.redis.rules.RedisAvailableTests;
+import org.springframework.integration.test.util.TestUtils;
+
+/**
+ * @author Gary Russell
+ * @since 4.0
+ *
+ */
+public class RedisLockRegistryTests extends RedisAvailableTests {
+
+ @Before
+ @After
+ public void shutDown() {
+ RedisTemplate template = this.createTemplate();
+ template.delete("rlrTests");
+ template.delete("rlrTests2");
+ }
+
+ private RedisTemplate createTemplate() {
+ RedisTemplate template = new RedisTemplate();
+ template.setConnectionFactory(this.getConnectionFactoryForTest());
+ template.setKeySerializer(new StringRedisSerializer());
+ template.afterPropertiesSet();
+ return template;
+ }
+
+ @Test
+ @RedisAvailable
+ public void testLock() throws Exception {
+ RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests");
+ for (int i = 0; i < 10; i++) {
+ Lock lock = registry.obtain("foo");
+ lock.lock();
+ try {
+ assertNotNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get());
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+ assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get());
+ }
+
+ @Test
+ @RedisAvailable
+ public void testLockInterruptibly() throws Exception {
+ RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests");
+ for (int i = 0; i < 10; i++) {
+ Lock lock = registry.obtain("foo");
+ lock.lockInterruptibly();
+ try {
+ assertNotNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get());
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+ assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get());
+ }
+
+ @Test
+ @RedisAvailable
+ public void testRentrantLock() throws Exception {
+ RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests");
+ for (int i = 0; i < 10; i++) {
+ Lock lock1 = registry.obtain("foo");
+ lock1.lock();
+ try {
+ Lock lock2 = registry.obtain("foo");
+ assertSame(lock1, lock2);
+ lock2.lock();
+ try {
+
+ }
+ finally {
+ lock2.unlock();
+ }
+ }
+ finally {
+ lock1.unlock();
+ }
+ }
+ assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get());
+ }
+
+ @Test
+ @RedisAvailable
+ public void testRentrantLockInterruptibly() throws Exception {
+ RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests");
+ for (int i = 0; i < 10; i++) {
+ Lock lock1 = registry.obtain("foo");
+ lock1.lockInterruptibly();
+ try {
+ Lock lock2 = registry.obtain("foo");
+ assertSame(lock1, lock2);
+ lock2.lockInterruptibly();
+ try {
+
+ }
+ finally {
+ lock2.unlock();
+ }
+ }
+ finally {
+ lock1.unlock();
+ }
+ }
+ assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get());
+ }
+
+ @Test
+ @RedisAvailable
+ public void testTwoLocks() throws Exception {
+ RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests");
+ for (int i = 0; i < 10; i++) {
+ Lock lock1 = registry.obtain("foo");
+ lock1.lockInterruptibly();
+ try {
+ Lock lock2 = registry.obtain("bar");
+ assertNotSame(lock1, lock2);
+ lock2.lockInterruptibly();
+ try {
+
+ }
+ finally {
+ lock2.unlock();
+ }
+ }
+ finally {
+ lock1.unlock();
+ }
+ }
+ assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get());
+ }
+
+ @Test
+ @RedisAvailable
+ public void testTwoThreadsSecondFailsToGetLock() throws Exception {
+ final RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests");
+ final Lock lock1 = registry.obtain("foo");
+ lock1.lockInterruptibly();
+ final AtomicBoolean locked = new AtomicBoolean();
+ final CountDownLatch latch = new CountDownLatch(1);
+ Future