diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/util/GemfireLockRegistry.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/util/GemfireLockRegistry.java
new file mode 100644
index 0000000000..a3605e9e2d
--- /dev/null
+++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/util/GemfireLockRegistry.java
@@ -0,0 +1,55 @@
+/*
+ * 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.gemfire.util;
+
+import java.util.concurrent.locks.Lock;
+
+import org.springframework.integration.support.locks.LockRegistry;
+import org.springframework.util.Assert;
+
+import com.gemstone.gemfire.cache.Cache;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.Scope;
+
+/**
+ * Implementation of {@link LockRegistry} providing a distributed lock using Gemfire.
+ *
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public class GemfireLockRegistry implements LockRegistry {
+
+ public static final String LOCK_REGISTRY_REGION = "LockRegistry";
+
+ private final Region region;
+
+ public GemfireLockRegistry(Cache cache) {
+ Assert.notNull(cache, "'cache' must not be null");
+ this.region = cache.createRegionFactory().setScope(Scope.GLOBAL).create(LOCK_REGISTRY_REGION);
+ }
+
+ public GemfireLockRegistry(Region region) {
+ Assert.notNull(region, "'region' must not be null");
+ this.region = region;
+ }
+
+ @Override
+ public Lock obtain(Object lockKey) {
+ return this.region.getDistributedLock(lockKey);
+ }
+
+}
diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/util/package-info.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/util/package-info.java
new file mode 100644
index 0000000000..fe86a3b804
--- /dev/null
+++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/util/package-info.java
@@ -0,0 +1,4 @@
+/**
+ * Provides utility classes.
+ */
+package org.springframework.integration.gemfire.util;
diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests-context.xml
new file mode 100644
index 0000000000..516f307764
--- /dev/null
+++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests-context.xml
@@ -0,0 +1,37 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests.java
new file mode 100644
index 0000000000..3973304223
--- /dev/null
+++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests.java
@@ -0,0 +1,186 @@
+/*
+ * 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.gemfire.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.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.aggregator.ReleaseStrategy;
+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
+ * @author Artem Bilan
+ * @since 4.0
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class AggregatorWithGemfireLocksTests {
+
+ @Autowired
+ private LatchingReleaseStrategy releaseStrategy;
+
+ @Autowired
+ private MessageChannel in;
+
+ @Autowired
+ private MessageChannel in2;
+
+ @Autowired
+ private PollableChannel out;
+
+ private volatile Exception exception;
+
+ @Test
+ 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));
+ this.releaseStrategy.latch1.countDown();
+ assertNotNull(this.out.receive(10000));
+ assertEquals(1, this.releaseStrategy.maxCallers.get());
+ assertNull("Unexpected exception:" + (this.exception != null ? this.exception.toString() : ""), this.exception);
+ }
+
+ @Test
+ 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));
+ 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());
+ assertNull("Unexpected exception:" + (this.exception != null ? this.exception.toString() : ""), this.exception);
+ }
+
+ @Test
+ 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));
+ this.releaseStrategy.latch1.countDown();
+ assertNotNull(this.out.receive(10000));
+ assertEquals(1, this.releaseStrategy.maxCallers.get());
+ assertNull("Unexpected exception:" + (this.exception != null ? this.exception.toString() : ""), this.exception);
+ }
+
+ 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;
+ }
+
+ 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/src/reference/docbook/gemfire.xml b/src/reference/docbook/gemfire.xml
index df9edb4100..f5abd70875 100644
--- a/src/reference/docbook/gemfire.xml
+++ b/src/reference/docbook/gemfire.xml
@@ -178,4 +178,26 @@ In the above examle, the cache and region are configured using the spring-gemfir
Note the pool element is configured with the address of a cache server (a locator may be substituted here). The region is configured as a 'PROXY' so that no data will be stored locally. The region's id corresponds to a region with the same name configured in the cache server.
+
+ Gemfire Lock Registry
+
+
+ Starting with version 4.0, the GemfireLockRegistry is
+ available. Certain components (for example aggregator and resequencer) use a lock obtained from
+ a LockRegistry instance to ensure
+ that only one thread is manipulating a group at a time. The DefaultLockRegistry
+ performs this function within a single component; you can now configure an external lock registry
+ on these components. When used with a shared MessageGroupStore,
+ the GemfireLockRegistry can be use to provide this functionality across
+ multiple application instances, such that only one instance can manipulate the group at a time.
+
+ One of the GemfireLockRegistry constructors requires a Region
+ as an argument; it is used to obtain a Lock via the
+ getDistributedLock() method. This operation requires GLOBAL
+ scope for the Region.
+ Another constructor requires Cache and the Region
+ will be created with GLOBAL scope and with the name LockRegistry.
+
+
+
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml
index 1cdfe9c7f1..c043b1eb93 100644
--- a/src/reference/docbook/whats-new.xml
+++ b/src/reference/docbook/whats-new.xml
@@ -133,13 +133,15 @@
For more information, see .
-
- RedisLockRegistry
+
+ RedisLockRegistry and GemfireLockRegistry
- The RedisLockRegistry is now available supporting global locks visible
- to multiple application instances/servers. This can be used with aggregating message handlers
- across multiple application instances such that group release will occur on only one instance.
- For more information, see and .
+ The RedisLockRegistry and GemfireLockRegistry are now
+ available supporting global locks visible to multiple application instances/servers.
+ These can be used with aggregating message handlers across multiple application instances such
+ that group release will occur on only one instance.
+ For more information, see ,
+ and .