INT-3359: Add GemfireLockRegistry
JIRA: https://jira.spring.io/browse/INT-3359
This commit is contained in:
Artem Bilan
committed by
Gary Russell
Gary Russell
parent
89a7743640
commit
93ad65e7ce
@@ -0,0 +1,37 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:int="http://www.springframework.org/schema/integration"
|
||||
xmlns:gfe="http://www.springframework.org/schema/gemfire"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
|
||||
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
|
||||
|
||||
<gfe:cache />
|
||||
|
||||
<bean id="lockRegistry" class="org.springframework.integration.gemfire.util.GemfireLockRegistry">
|
||||
<constructor-arg ref="gemfireCache"/>
|
||||
</bean>
|
||||
|
||||
<bean id="lockRegistry2" class="org.springframework.integration.gemfire.util.GemfireLockRegistry">
|
||||
<constructor-arg value="#{gemfireCache.getRegion(T(org.springframework.integration.gemfire.util.GemfireLockRegistry).LOCK_REGISTRY_REGION)}"/>
|
||||
</bean>
|
||||
|
||||
<bean id="latching"
|
||||
class="org.springframework.integration.gemfire.util.AggregatorWithGemfireLocksTests$LatchingReleaseStrategy"/>
|
||||
|
||||
<bean id="sms" class="org.springframework.integration.store.SimpleMessageStore"/>
|
||||
|
||||
<int:aggregator input-channel="in" release-strategy="latching" output-channel="out"
|
||||
message-store="sms"
|
||||
expire-groups-upon-completion="true" lock-registry="lockRegistry"/>
|
||||
|
||||
<int:aggregator input-channel="in2" release-strategy="latching" output-channel="out"
|
||||
message-store="sms"
|
||||
expire-groups-upon-completion="true" lock-registry="lockRegistry2"/>
|
||||
|
||||
<int:channel id="out">
|
||||
<int:queue/>
|
||||
</int:channel>
|
||||
|
||||
</beans>
|
||||
@@ -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<String>("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<String>(payload, stubHeaders(sequence, 2, correlation)));
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
exception = e;
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private Map<String, Object> stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) {
|
||||
Map<String, Object> headers = new HashMap<String, Object>();
|
||||
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();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user