Another fix for failing tests

* `JmsOutboundGatewayTests`: just increase timeouts. Looks like embedded ActiveMQ Broker takes more time to interact on high-loaded builds
* `TcpOutboundGatewayTests`: Rework the race condition fix to the atomic `remoteTimeout`  change using mocks
This commit is contained in:
Artem Bilan
2015-10-21 19:04:01 -04:00
parent aa3e3c7098
commit d0af90db62
2 changed files with 54 additions and 27 deletions

View File

@@ -57,14 +57,19 @@ import org.apache.commons.logging.LogFactory;
import org.apache.log4j.Level; import org.apache.log4j.Level;
import org.junit.Rule; import org.junit.Rule;
import org.junit.Test; import org.junit.Test;
import org.mockito.Matchers;
import org.mockito.Mockito; import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.core.serializer.DefaultSerializer; import org.springframework.core.serializer.DefaultSerializer;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.FailoverClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.FailoverClientConnectionFactory;
@@ -383,26 +388,41 @@ public class TcpOutboundGatewayTests {
QueueChannel replyChannel = new QueueChannel(); QueueChannel replyChannel = new QueueChannel();
gateway.setRequiresReply(true); gateway.setRequiresReply(true);
gateway.setOutputChannel(replyChannel); gateway.setOutputChannel(replyChannel);
gateway.setRemoteTimeout(500);
ValueExpression<Long> remoteTimeoutExpression = Mockito.spy(new ValueExpression<Long>(500L));
final AtomicBoolean remoteTimeoutUsed = new AtomicBoolean();
Mockito.doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
if (!remoteTimeoutUsed.getAndSet(true)) {
// increase the timeout after the first send
gateway.setRemoteTimeout(5000);
}
return invocation.callRealMethod();
}
}).when(remoteTimeoutExpression)
.getValue(Mockito.any(EvaluationContext.class), Matchers.any());
gateway.setRemoteTimeoutExpression(remoteTimeoutExpression);
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
Future<Integer>[] results = (Future<Integer>[]) new Future<?>[2]; Future<Integer>[] results = (Future<Integer>[]) new Future<?>[2];
final CountDownLatch secondMessageLatch = new CountDownLatch(1);
for (int i = 0; i < 2; i++) { for (int i = 0; i < 2; i++) {
final int j = i; final int j = i;
results[j] = (Executors.newSingleThreadExecutor().submit(new Callable<Integer>() { results[j] = (Executors.newSingleThreadExecutor().submit(new Callable<Integer>() {
@Override @Override
public Integer call() throws Exception { public Integer call() throws Exception {
try { gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
}
finally {
// increase the timeout after the first send
if (j > 0) {
gateway.setRemoteTimeout(5000);
}
}
return j; return j;
} }
})); }));
} }
// wait until the server side has processed both requests // wait until the server side has processed both requests
assertTrue(serverLatch.await(10, TimeUnit.SECONDS)); assertTrue(serverLatch.await(10, TimeUnit.SECONDS));

View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.integration.jms; package org.springframework.integration.jms;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
@@ -64,8 +65,8 @@ import org.springframework.util.ObjectUtils;
/** /**
* @author Gary Russell * @author Gary Russell
* @author Artem Bilan
* @since 2.2.4 * @since 2.2.4
*
*/ */
public class JmsOutboundGatewayTests { public class JmsOutboundGatewayTests {
@@ -81,8 +82,8 @@ public class JmsOutboundGatewayTests {
gateway.setBeanFactory(mock(BeanFactory.class)); gateway.setBeanFactory(mock(BeanFactory.class));
gateway.afterPropertiesSet(); gateway.afterPropertiesSet();
assertEquals("JMS_OutboundGateway@" + ObjectUtils.getIdentityHexString(gateway) + assertEquals("JMS_OutboundGateway@" + ObjectUtils.getIdentityHexString(gateway) +
".replyListener", ".replyListener",
TestUtils.getPropertyValue(gateway, "replyContainer.beanName")); TestUtils.getPropertyValue(gateway, "replyContainer.beanName"));
} }
@Test @Test
@@ -94,15 +95,17 @@ public class JmsOutboundGatewayTests {
gateway.setUseReplyContainer(true); gateway.setUseReplyContainer(true);
ReplyContainerProperties replyContainerProperties = new ReplyContainerProperties(); ReplyContainerProperties replyContainerProperties = new ReplyContainerProperties();
final List<Throwable> errors = new ArrayList<Throwable>(); final List<Throwable> errors = new ArrayList<Throwable>();
ErrorHandlingTaskExecutor errorHandlingTaskExecutor = new ErrorHandlingTaskExecutor(Executors.newFixedThreadPool(10), new ErrorHandler() { ErrorHandlingTaskExecutor errorHandlingTaskExecutor =
new ErrorHandlingTaskExecutor(Executors.newFixedThreadPool(10), new ErrorHandler() {
@Override @Override
public void handleError(Throwable t) { public void handleError(Throwable t) {
logger.info("Error:", t); logger.info("Error:", t);
errors.add(t); errors.add(t);
throw new RuntimeException(t); throw new RuntimeException(t);
} }
});
});
replyContainerProperties.setTaskExecutor(errorHandlingTaskExecutor); replyContainerProperties.setTaskExecutor(errorHandlingTaskExecutor);
replyContainerProperties.setRecoveryInterval(100L); replyContainerProperties.setRecoveryInterval(100L);
gateway.setReplyContainerProperties(replyContainerProperties); gateway.setReplyContainerProperties(replyContainerProperties);
@@ -115,7 +118,9 @@ public class JmsOutboundGatewayTests {
public Connection answer(InvocationOnMock invocation) throws Throwable { public Connection answer(InvocationOnMock invocation) throws Throwable {
int theCount = connectionAttempts.incrementAndGet(); int theCount = connectionAttempts.incrementAndGet();
if (theCount > 1 && theCount < 4) { if (theCount > 1 && theCount < 4) {
throw new JmsException("bar") {}; throw new JmsException("bar") {
};
} }
return connection; return connection;
} }
@@ -134,7 +139,9 @@ public class JmsOutboundGatewayTests {
public Message answer(InvocationOnMock invocation) throws Throwable { public Message answer(InvocationOnMock invocation) throws Throwable {
int theCount = count.incrementAndGet(); int theCount = count.incrementAndGet();
if (theCount > 1 && theCount < 4) { if (theCount > 1 && theCount < 4) {
throw new JmsException("foo") {}; throw new JmsException("foo") {
};
} }
if (theCount > 4) { if (theCount > 4) {
Thread.sleep(100); Thread.sleep(100);
@@ -190,7 +197,7 @@ public class JmsOutboundGatewayTests {
CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory( CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")); new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
JmsTemplate template = new JmsTemplate(connectionFactory2); JmsTemplate template = new JmsTemplate(connectionFactory2);
template.setReceiveTimeout(5000); template.setReceiveTimeout(10000);
template.afterPropertiesSet(); template.afterPropertiesSet();
final Message request = template.receive(requestQ); final Message request = template.receive(requestQ);
assertNotNull(request); assertNotNull(request);
@@ -205,7 +212,7 @@ public class JmsOutboundGatewayTests {
} }
}; };
template.send(replyQ, reply); template.send(replyQ, reply);
org.springframework.messaging.Message<?> received = queueChannel.receive(10000); org.springframework.messaging.Message<?> received = queueChannel.receive(20000);
assertNotNull(received); assertNotNull(received);
assertEquals("bar", received.getPayload()); assertEquals("bar", received.getPayload());
gateway.stop(); gateway.stop();
@@ -240,7 +247,7 @@ public class JmsOutboundGatewayTests {
CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory( CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")); new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
JmsTemplate template = new JmsTemplate(connectionFactory2); JmsTemplate template = new JmsTemplate(connectionFactory2);
template.setReceiveTimeout(5000); template.setReceiveTimeout(10000);
template.afterPropertiesSet(); template.afterPropertiesSet();
final Message request = template.receive(requestQ); final Message request = template.receive(requestQ);
assertNotNull(request); assertNotNull(request);
@@ -255,7 +262,7 @@ public class JmsOutboundGatewayTests {
} }
}; };
template.send(replyQ, reply); template.send(replyQ, reply);
org.springframework.messaging.Message<?> received = queueChannel.receive(10000); org.springframework.messaging.Message<?> received = queueChannel.receive(20000);
assertNotNull(received); assertNotNull(received);
assertEquals("bar", received.getPayload()); assertEquals("bar", received.getPayload());
connectionFactory1.destroy(); connectionFactory1.destroy();