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

Fix logic in the PollingTransactionTests

Browse Source 
The `transactionWithCommitAndAdvices()` uses `SimpleRepeatAdvice`
which essentially performs `doPoll()` twice.
In this case we don't have enough messages in the `goodInputWithAdvice`
queue

* Add one more message to the `goodInputWithAdvice` queue during testing
* Increase all the timeouts
* Increase capacity in queues for the `transactionWithCommitAndAdvices()`

**Cherry-pick to 5.0.x and 4.3.x**
This commit is contained in:
Artem Bilan
2018-04-26 09:22:00 -04:00
parent 0ef659b8a7
commit 000f2972f4
2 changed files with 72 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-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.
@@ -34,6 +34,7 @@ import org.junit.Test;
import org.springframework.aop.Advisor;
import org.springframework.aop.ProxyMethodInvocation;
import org.springframework.aop.framework.Advised;
import org.springframework.context.Lifecycle;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.test.util.TestUtils;
@@ -68,8 +69,9 @@ public class PollingTransactionTests {
PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount());
assertEquals(0, txManager.getRollbackCount());
input.send(new GenericMessage<String>("test"));
txManager.waitForCompletion(1000);
context.getBean("goodService", Lifecycle.class).start();
input.send(new GenericMessage<>("test"));
txManager.waitForCompletion(10000);
Message<?> message = output.receive(0);
assertNotNull(message);
assertEquals(1, txManager.getCommitCount());
@@ -82,11 +84,12 @@ public class PollingTransactionTests {
public void transactionWithCommitAndAdvices() throws InterruptedException {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"transactionTests.xml", this.getClass());
PollingConsumer advicedPoller = context.getBean("advicedSa", PollingConsumer.class);
PollingConsumer advisedPoller = context.getBean("advisedSa", PollingConsumer.class);
List<Advice> adviceChain = TestUtils.getPropertyValue(advicedPoller, "adviceChain", List.class);
List<Advice> adviceChain = TestUtils.getPropertyValue(advisedPoller, "adviceChain", List.class);
assertEquals(4, adviceChain.size());
Runnable poller = TestUtils.getPropertyValue(advicedPoller, "poller", Runnable.class);
advisedPoller.start();
Runnable poller = TestUtils.getPropertyValue(advisedPoller, "poller", Runnable.class);
Callable<?> pollingTask = TestUtils.getPropertyValue(poller, "pollingTask", Callable.class);
assertTrue("Poller is not Advised", pollingTask instanceof Advised);
Advisor[] advisors = ((Advised) pollingTask).getAdvisors();
@@ -98,10 +101,14 @@ public class PollingTransactionTests {
PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount());
assertEquals(0, txManager.getRollbackCount());
input.send(new GenericMessage<>("test"));
input.send(new GenericMessage<>("test2"));
txManager.waitForCompletion(10000);
Message<?> message = output.receive(0);
assertNotNull(message);
message = output.receive(0);
assertNotNull(message);
assertEquals(0, txManager.getRollbackCount());
context.close();
}
@@ -115,8 +122,11 @@ public class PollingTransactionTests {
PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount());
assertEquals(0, txManager.getRollbackCount());
input.send(new GenericMessage<String>("test"));
txManager.waitForCompletion(1000);
context.getBean("badService", Lifecycle.class).start();
input.send(new GenericMessage<>("test"));
txManager.waitForCompletion(10000);
Message<?> message = output.receive(0);
assertNull(message);
assertEquals(0, txManager.getCommitCount());
@@ -132,10 +142,10 @@ public class PollingTransactionTests {
PollableChannel input = (PollableChannel) context.getBean("input");
PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount());
input.send(new GenericMessage<String>("test"));
Message<?> reply = output.receive(3000);
input.send(new GenericMessage<>("test"));
Message<?> reply = output.receive(10000);
assertNotNull(reply);
txManager.waitForCompletion(3000);
txManager.waitForCompletion(10000);
assertEquals(1, txManager.getCommitCount());
assertEquals(Propagation.REQUIRED.value(), txManager.getLastDefinition().getPropagationBehavior());
context.close();
@@ -149,25 +159,25 @@ public class PollingTransactionTests {
PollableChannel input = (PollableChannel) context.getBean("input");
PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount());
input.send(new GenericMessage<String>("test"));
Message<?> reply = output.receive(3000);
input.send(new GenericMessage<>("test"));
Message<?> reply = output.receive(10000);
assertNotNull(reply);
txManager.waitForCompletion(3000);
txManager.waitForCompletion(10000);
assertEquals(1, txManager.getCommitCount());
assertEquals(Propagation.REQUIRES_NEW.value(), txManager.getLastDefinition().getPropagationBehavior());
context.close();
}
@Test
public void propagationSupports() throws InterruptedException {
public void propagationSupports() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"propagationSupportsTests.xml", this.getClass());
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
PollableChannel input = (PollableChannel) context.getBean("input");
PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount());
input.send(new GenericMessage<String>("test"));
Message<?> reply = output.receive(3000);
input.send(new GenericMessage<>("test"));
Message<?> reply = output.receive(10000);
assertNotNull(reply);
assertEquals(0, txManager.getCommitCount());
assertNull(txManager.getLastDefinition());
@@ -175,15 +185,15 @@ public class PollingTransactionTests {
}
@Test
public void propagationNotSupported() throws InterruptedException {
public void propagationNotSupported() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"propagationNotSupportedTests.xml", this.getClass());
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
PollableChannel input = (PollableChannel) context.getBean("input");
PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount());
input.send(new GenericMessage<String>("test"));
Message<?> reply = output.receive(3000);
input.send(new GenericMessage<>("test"));
Message<?> reply = output.receive(10000);
assertNotNull(reply);
assertEquals(0, txManager.getCommitCount());
assertNull(txManager.getLastDefinition());
@@ -191,7 +201,7 @@ public class PollingTransactionTests {
}
@Test
public void propagationMandatory() throws Throwable {
public void propagationMandatory() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"propagationMandatoryTests.xml", this.getClass());
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
@@ -199,8 +209,8 @@ public class PollingTransactionTests {
PollableChannel output = (PollableChannel) context.getBean("output");
PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel");
assertEquals(0, txManager.getCommitCount());
input.send(new GenericMessage<String>("test"));
Message<?> errorMessage = errorChannel.receive(3000);
input.send(new GenericMessage<>("test"));
Message<?> errorMessage = errorChannel.receive(10000);
assertNotNull(errorMessage);
Object payload = errorMessage.getPayload();
assertEquals(MessagingException.class, payload.getClass());
@@ -212,7 +222,7 @@ public class PollingTransactionTests {
}
@Test
public void commitFailureAndHandlerFailureTest() throws Throwable {
public void commitFailureAndHandlerFailureTest() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"transactionFailureTests.xml", this.getClass());
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManagerBad");
@@ -222,7 +232,7 @@ public class PollingTransactionTests {
PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel");
assertEquals(0, txManager.getCommitCount());
inputTxFail.send(new GenericMessage<>("commitFailureTest"));
Message<?> errorMessage = errorChannel.receive(10000);
Message<?> errorMessage = errorChannel.receive(20000);
assertNotNull(errorMessage);
Object payload = errorMessage.getPayload();
assertEquals(MessagingException.class, payload.getClass());
@@ -232,7 +242,7 @@ public class PollingTransactionTests {
assertNotNull(output.receive(0));
assertEquals(0, txManager.getCommitCount());
inputHandlerFail.send(new GenericMessage<>("handlerFalilureTest"));
inputHandlerFail.send(new GenericMessage<>("handlerFailureTest"));
errorMessage = errorChannel.receive(10000);
assertNotNull(errorMessage);
payload = errorMessage.getPayload();

62
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/transactionTests.xml
View File

@@ -1,53 +1,64 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="badInput">
<queue capacity="1" />
<queue capacity="1"/>
</channel>
<channel id="goodInput">
<queue capacity="1" />
<queue capacity="1"/>
</channel>
<channel id="goodInputWithAdvice">
<queue capacity="1" />
<queue capacity="2"/>
</channel>
<channel id="output">
<queue capacity="1" />
<queue capacity="2"/>
</channel>
<service-activator input-channel="badInput" ref="testBean"
method="bad" output-channel="output">
<service-activator id="badService"
auto-startup="false"
input-channel="badInput"
ref="testBean" method="bad"
output-channel="output">
<poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
<transactional transaction-manager="txManager" />
<transactional transaction-manager="txManager"/>
</poller>
</service-activator>
<service-activator input-channel="goodInput" ref="testBean"
method="good" output-channel="output">
<service-activator id="goodService"
auto-startup="false"
input-channel="goodInput"
ref="testBean" method="good"
output-channel="output">
<poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
<transactional transaction-manager="txManager" />
<transactional transaction-manager="txManager"/>
</poller>
</service-activator>
<service-activator id="advicedSa" input-channel="goodInputWithAdvice" ref="testBean"
method="good" output-channel="output">
<service-activator id="advisedSa"
auto-startup="false"
input-channel="goodInputWithAdvice"
ref="testBean" method="good"
output-channel="output">
<poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
<advice-chain>
<ref bean="txAdvise"/>
<ref bean="adviceA" />
<beans:bean class="org.springframework.integration.dispatcher.PollingTransactionTests.SampleAdvice"/>
<beans:bean class="org.springframework.integration.dispatcher.PollingTransactionTests.SimpleRepeatAdvice"/>
<ref bean="adviceA"/>
<beans:bean class="org.springframework.integration.dispatcher.PollingTransactionTests$SampleAdvice"/>
<beans:bean
class="org.springframework.integration.dispatcher.PollingTransactionTests$SimpleRepeatAdvice"/>
</advice-chain>
</poller>
</service-activator>
<beans:bean id="txAdvise" class="org.springframework.transaction.interceptor.TransactionInterceptor">
<beans:property name="transactionManager" ref="txManager"/>
<beans:property name="transactionAttributeSource">
@@ -58,12 +69,11 @@
</beans:bean>
</beans:property>
</beans:bean>
<beans:bean id="adviceA" class="org.springframework.integration.dispatcher.PollingTransactionTests.SampleAdvice"/>
<beans:bean id="testBean"
class="org.springframework.integration.dispatcher.TestBean" />
<beans:bean id="adviceA" class="org.springframework.integration.dispatcher.PollingTransactionTests$SampleAdvice"/>
<beans:bean id="testBean" class="org.springframework.integration.dispatcher.TestBean"/>
<beans:bean id="txManager" class="org.springframework.integration.util.TestTransactionManager"/>
<beans:bean id="txManager"
class="org.springframework.integration.util.TestTransactionManager" />
</beans:beans>