Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2291

Browse Source 
polished MixedDispatcherConfigurationScenarioTests.failoverNoLoadBalancingWithExecutorConcurrent() test even more, introduced 'taskSubmissionMonitor' as CountDownLatch to ensure that shutdown command is not called untill all tasks are submitted. Also made sure that ExecutorChannel is using the same tasks executor so shutdown command actually shuts down the same executor.
This commit is contained in:
Oleg Zhurakousky
2011-12-19 09:22:47 -05:00
parent 22cbe9e0dd
commit 8869da7a95
2 changed files with 29 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests-context.xml
View File

@@ -28,9 +28,7 @@
<dispatcher load-balancer="none" failover="true" task-executor="taskExecutor"/>
</channel>
<beans:bean id="taskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor">
<beans:property name="threadGroupName" value="dispatchers"/>
</beans:bean>
<beans:bean id="taskExecutor" class="java.util.concurrent.Executors" factory-method="newCachedThreadPool"/>
<!-- send async errors to null channel to avoid chatter -->
<bridge input-channel="errorChannel" output-channel="nullChannel"/>

52
spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.channel;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -61,7 +60,7 @@ public class MixedDispatcherConfigurationScenarioTests {
private static final int TOTAL_EXECUTIONS = 40;
private ExecutorService scheduler;
private ExecutorService executor;
private CountDownLatch allDone;
private CountDownLatch start;
@@ -91,9 +90,10 @@ public class MixedDispatcherConfigurationScenarioTests {
Mockito.reset(handlerA);
Mockito.reset(handlerB);
Mockito.reset(handlerC);
scheduler = Executors.newCachedThreadPool();
ac = new ClassPathXmlApplicationContext("MixedDispatcherConfigurationScenarioTests-context.xml",
MixedDispatcherConfigurationScenarioTests.class);
executor = ac.getBean("taskExecutor", ExecutorService.class);
allDone = new CountDownLatch(TOTAL_EXECUTIONS);
start = new CountDownLatch(1);
failed = new AtomicBoolean(false);
@@ -146,13 +146,13 @@ public class MixedDispatcherConfigurationScenarioTests {
}
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
executor.execute(messageSenderTask);
}
start.countDown();
allDone.await();
scheduler.shutdown();
scheduler.awaitTermination(5, TimeUnit.SECONDS);
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);
assertTrue("not all messages were accepted", failed.get());
verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message);
@@ -196,13 +196,13 @@ public class MixedDispatcherConfigurationScenarioTests {
}
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
executor.execute(messageSenderTask);
}
start.countDown();
allDone.await();
scheduler.shutdown();
scheduler.awaitTermination(5, TimeUnit.SECONDS);
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);
assertTrue("not all messages were accepted", failed.get());
verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message);
@@ -274,13 +274,13 @@ public class MixedDispatcherConfigurationScenarioTests {
}
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
executor.execute(messageSenderTask);
}
start.countDown();
allDone.await();
scheduler.shutdown();
scheduler.awaitTermination(5, TimeUnit.SECONDS);
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);
assertTrue("not all messages were accepted", failed.get());
verify(handlerA, times(14)).handleMessage(message);
@@ -334,13 +334,13 @@ public class MixedDispatcherConfigurationScenarioTests {
}
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
executor.execute(messageSenderTask);
}
start.countDown();
allDone.await();
scheduler.shutdown();
scheduler.awaitTermination(5, TimeUnit.SECONDS);
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);
assertTrue("not all messages were accepted", failed.get());
verify(handlerA, times(14)).handleMessage(message);
@@ -413,13 +413,13 @@ public class MixedDispatcherConfigurationScenarioTests {
}
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
executor.execute(messageSenderTask);
}
start.countDown();
allDone.await();
scheduler.shutdown();
scheduler.awaitTermination(5, TimeUnit.SECONDS);
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);
assertFalse("not all messages were accepted", failed.get());
verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message);
@@ -434,7 +434,9 @@ public class MixedDispatcherConfigurationScenarioTests {
final UnicastingDispatcher dispatcher = channel.getDispatcher();
dispatcher.addHandler(handlerA);
dispatcher.addHandler(handlerB);
dispatcher.addHandler(handlerC);
dispatcher.addHandler(handlerC);
final CountDownLatch taskSubmissionMonitor = new CountDownLatch(TOTAL_EXECUTIONS);
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) {
@@ -460,23 +462,25 @@ public class MixedDispatcherConfigurationScenarioTests {
Runnable messageSenderTask = new Runnable() {
public void run() {
try {
start.await();
start.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
channel.send(message);
taskSubmissionMonitor.countDown();
}
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
executor.execute(messageSenderTask);
}
start.countDown();
allDone.await();
Thread.sleep(500);
taskSubmissionMonitor.await();
scheduler.shutdown();
scheduler.awaitTermination(5, TimeUnit.SECONDS);
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);
verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message);
verify(handlerB, times(TOTAL_EXECUTIONS)).handleMessage(message);