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

Fixes for Sporadic Test Failures

Browse Source 
* `ClientWebSocketContainer`: add some synchronization to avoid race conditions: https://build.spring.io/browse/INT-B41-492
* `TomcatWebSocketTestServer`: convert to `0` port to rely on the OS resolution for `localPort`
* Add `LogAdjustingTestSupport` for STOMP test
* `SftpServerTests`: use `0` port to rely on the OS resolution for `localPort`
* `ImapMailReceiver`, `OutboundGatewayFunctionTests` (JMS), `CachingClientConnectionFactoryTests`,
`AsyncGatewayTests`, `AsyncMessagingTemplateTests`, `GatewayParserTests`, `PriorityChannelTests`, `AggregatorIntegrationTests`: increase timeout
* `EnableIntegrationTests`: use `LogAdjustingTestSupport`
* `FileOutboundChannelAdapterParserTests`: rework `Thread.sleep()` with `CountDownLatch`
* `ConnectionToConnectionTests`: increase timeout and count iteration. Previously with `1sec` we may lose some events. And we can't just rely on the `10sec`,
because the last iteration will be so long
* `TcpOutboundGatewayTests`: `500ms` is so big timeout to wait for the `Exception` that in the high load environment we can yield to other Thread so long.
Like in our case to `server` Thread to send the reply for us. Therefore decrease the Exception timeout to the `50ms` and increase server delay to `2sec`
* `StompInboundChannelAdapterWebSocketIntegrationTests`: remove `@Qualifier("taskScheduler")` as a potential candidate to test against latest SF changes.
We're fine with `SF-4.2.2` and it is just a test-case. So, I don't see reason to wait for their fix here.

STOMP: `session = null` in adapters for any transportError

Polishing
This commit is contained in:
Artem Bilan
2015-11-13 22:39:00 -05:00
committed by Gary Russell
parent e45bb36be2
commit 9d8e2b61f6
23 changed files with 150 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java
View File

@@ -158,7 +158,7 @@ public class AggregatorIntegrationTests {
Thread.sleep(100);
}
assertTrue("Group did not complete", n < 100);
assertNotNull(this.output.receive(1000));
assertNotNull(this.output.receive(10000));
assertNull(this.discard.receive(0));
}
}
@@ -183,7 +183,7 @@ public class AggregatorIntegrationTests {
}
}
assertTrue("Group did not complete", n < 100);
Message<?> receive = this.output.receive(1000);
Message<?> receive = this.output.receive(10000);
assertNotNull(receive);
assertEquals(Collections.singletonList(1), receive.getPayload());
assertNull(this.discard.receive(0));
@@ -199,7 +199,7 @@ public class AggregatorIntegrationTests {
// As far as 'group.size() >= 2' it will be scheduled to 'forceComplete'
this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(2, stubHeaders(2, 6, 1)));
assertNull(this.output.receive(0));
Message<?> receive = this.output.receive(500);
Message<?> receive = this.output.receive(10000);
assertNotNull(receive);
assertEquals(2, ((Collection<?>) receive.getPayload()).size());
assertNull(this.discard.receive(0));
@@ -215,14 +215,14 @@ public class AggregatorIntegrationTests {
this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(5, stubHeaders(5, 6, 1)));
assertNull(this.output.receive(0));
receive = this.output.receive(500);
receive = this.output.receive(10000);
assertNotNull(receive);
assertEquals(3, ((Collection<?>) receive.getPayload()).size());
assertNull(this.discard.receive(0));
// The last message in the sequence - normal release by provided 'ReleaseStrategy'
this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(6, stubHeaders(6, 6, 1)));
receive = this.output.receive(500);
receive = this.output.receive(10000);
assertNotNull(receive);
assertEquals(1, ((Collection<?>) receive.getPayload()).size());
assertNull(this.discard.receive(0));

26
spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java
View File

@@ -25,6 +25,7 @@ import static org.junit.Assert.assertTrue;
import java.util.Comparator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -236,22 +237,23 @@ public class PriorityChannelTests {
public void testTimeoutElapses() throws InterruptedException {
final PriorityChannel channel = new PriorityChannel(1);
final AtomicBoolean sentSecondMessage = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Executor executor = Executors.newSingleThreadScheduledExecutor();
ExecutorService executor = Executors.newSingleThreadScheduledExecutor();
channel.send(new GenericMessage<String>("test-1"));
executor.execute(new Runnable() {
@Override
public void run() {
sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), 10));
latch.countDown();
}
});
assertFalse(sentSecondMessage.get());
Thread.sleep(1000);
Message<?> message1 = channel.receive();
executor.shutdown();
assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
Message<?> message1 = channel.receive(10000);
assertNotNull(message1);
assertEquals("test-1", message1.getPayload());
latch.await(10000, TimeUnit.MILLISECONDS);
assertFalse(sentSecondMessage.get());
assertNull(channel.receive(0));
}
@@ -286,22 +288,21 @@ public class PriorityChannelTests {
public void testIndefiniteTimeout() throws InterruptedException {
final PriorityChannel channel = new PriorityChannel(1);
final AtomicBoolean sentSecondMessage = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Executor executor = Executors.newSingleThreadScheduledExecutor();
ExecutorService executor = Executors.newSingleThreadScheduledExecutor();
channel.send(new GenericMessage<String>("test-1"));
executor.execute(new Runnable() {
@Override
public void run() {
sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), -1));
latch.countDown();
}
});
assertFalse(sentSecondMessage.get());
Thread.sleep(500);
Message<?> message1 = channel.receive();
Message<?> message1 = channel.receive(1000);
assertNotNull(message1);
assertEquals("test-1", message1.getPayload());
latch.await(1000, TimeUnit.MILLISECONDS);
executor.shutdown();
assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
assertTrue(sentSecondMessage.get());
Message<?> message2 = channel.receive();
assertNotNull(message2);
@@ -322,9 +323,11 @@ public class PriorityChannelTests {
String s2 = (String) message2.getPayload();
return s1.compareTo(s2);
}
}
public static class FooHeaderComparator implements Comparator<Message<?>> {
@Override
public int compare(Message<?> message1, Message<?> message2) {
Integer foo1 = (Integer) message1.getHeaders().get("foo");
@@ -333,6 +336,7 @@ public class PriorityChannelTests {
foo2 = foo2 != null ? foo2 : 0;
return foo2.compareTo(foo1);
}
}
}

5
spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java
View File

@@ -36,6 +36,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.junit.runner.RunWith;
import reactor.rx.Promise;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanNameAware;
@@ -64,8 +65,6 @@ import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import reactor.rx.Promise;
/**
* @author Mark Fisher
* @author Artem Bilan
@@ -196,7 +195,7 @@ public class GatewayParserTests {
this.startResponder(requestChannel, replyChannel);
TestService service = context.getBean("asyncCompletable", TestService.class);
CompletableFuture<String> result = service.completable("foo").thenApply(String::toUpperCase);
String reply = result.get(1, TimeUnit.SECONDS);
String reply = result.get(10, TimeUnit.SECONDS);
assertEquals("FOO", reply);
assertThat(thread.get().getName(), startsWith("testExec-"));
assertNotNull(TestUtils.getPropertyValue(context.getBean("&asyncCompletable"), "asyncExecutor"));

4
spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
View File

@@ -262,6 +262,10 @@ public class EnableIntegrationTests extends LogAdjustingTestSupport {
@Qualifier("enableIntegrationTests.ContextConfiguration2.sendAsyncHandler.serviceActivator")
private AbstractEndpoint sendAsyncHandler;
public EnableIntegrationTests() {
super("org.springframework.integration", "org.springframework");
}
@Test
public void testAnnotatedServiceActivator() {
assertEquals(10L, TestUtils.getPropertyValue(this.serviceActivatorEndpoint, "maxMessagesPerPoll"));

25
spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java
View File

@@ -58,7 +58,7 @@ public class AsyncMessagingTemplateTests {
template.setDefaultDestination(channel);
Message<?> message = MessageBuilder.withPayload("test").build();
Future<?> future = template.asyncSend(message);
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
assertNull(future.get(10000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals(message, result);
}
@@ -69,7 +69,7 @@ public class AsyncMessagingTemplateTests {
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
Message<?> message = MessageBuilder.withPayload("test").build();
Future<?> future = template.asyncSend(channel, message);
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
assertNull(future.get(10000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals(message, result);
}
@@ -84,7 +84,7 @@ public class AsyncMessagingTemplateTests {
template.setBeanFactory(context);
Message<?> message = MessageBuilder.withPayload("test").build();
Future<?> future = template.asyncSend("testChannel", message);
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
assertNull(future.get(10000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals(message, result);
}
@@ -104,7 +104,7 @@ public class AsyncMessagingTemplateTests {
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
template.setDefaultDestination(channel);
Future<?> future = template.asyncConvertAndSend("test");
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
assertNull(future.get(10000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals("test", result.getPayload());
}
@@ -114,7 +114,7 @@ public class AsyncMessagingTemplateTests {
QueueChannel channel = new QueueChannel();
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
Future<?> future = template.asyncConvertAndSend(channel, "test");
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
assertNull(future.get(10000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals("test", result.getPayload());
}
@@ -128,7 +128,7 @@ public class AsyncMessagingTemplateTests {
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
template.setBeanFactory(context);
Future<?> future = template.asyncConvertAndSend("testChannel", "test");
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
assertNull(future.get(10000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals("test", result.getPayload());
}
@@ -150,7 +150,7 @@ public class AsyncMessagingTemplateTests {
Future<Message<?>> result = template.asyncReceive();
sendMessageAfterDelay(channel, new GenericMessage<String>("test"), 200);
long start = System.currentTimeMillis();
assertNotNull(result.get(10000, TimeUnit.MILLISECONDS));
assertNotNull(result.get(100000, TimeUnit.MILLISECONDS));
long elapsed = System.currentTimeMillis() - start;
assertEquals("test", result.get().getPayload());
assertTrue(elapsed >= 200-safety);
@@ -163,7 +163,7 @@ public class AsyncMessagingTemplateTests {
Future<Message<?>> result = template.asyncReceive(channel);
sendMessageAfterDelay(channel, new GenericMessage<String>("test"), 200);
long start = System.currentTimeMillis();
assertNotNull(result.get(1000, TimeUnit.MILLISECONDS));
assertNotNull(result.get(10000, TimeUnit.MILLISECONDS));
long elapsed = System.currentTimeMillis() - start;
assertEquals("test", result.get().getPayload());
assertTrue(elapsed >= 200-safety);
@@ -348,7 +348,7 @@ public class AsyncMessagingTemplateTests {
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
template.setDefaultDestination(channel);
long start = System.currentTimeMillis();
Future<String> result = template.asyncConvertSendAndReceive(new Integer(123), new TestMessagePostProcessor());
Future<String> result = template.asyncConvertSendAndReceive(123, new TestMessagePostProcessor());
assertNotNull(result.get());
long elapsed = System.currentTimeMillis() - start;
@@ -435,8 +435,10 @@ public class AsyncMessagingTemplateTests {
}
private static void sendMessageAfterDelay(final MessageChannel channel, final GenericMessage<String> message, final int delay) {
private static void sendMessageAfterDelay(final MessageChannel channel, final GenericMessage<String> message,
final int delay) {
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
Thread.sleep(delay);
@@ -447,6 +449,7 @@ public class AsyncMessagingTemplateTests {
}
channel.send(message);
}
});
}
@@ -480,6 +483,7 @@ public class AsyncMessagingTemplateTests {
String header = requestMessage.getHeaders().get("foo", String.class);
return (header != null) ? result + "-" + header : result;
}
}
@@ -488,6 +492,7 @@ public class AsyncMessagingTemplateTests {
public Message<?> postProcessMessage(Message<?> message) {
return MessageBuilder.fromMessage(message).setHeader("foo", "bar").build();
}
}
}

27
spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java
View File

@@ -33,6 +33,9 @@ import java.util.concurrent.atomic.AtomicReference;
import org.hamcrest.Matchers;
import org.junit.Test;
import reactor.Environment;
import reactor.fn.Consumer;
import reactor.rx.Promise;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.annotation.Gateway;
@@ -47,10 +50,6 @@ import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import reactor.Environment;
import reactor.rx.Promise;
import reactor.fn.Consumer;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
@@ -82,7 +81,7 @@ public class AsyncGatewayTests {
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Future<Message<?>> f = service.returnMessage("foo");
long start = System.currentTimeMillis();
Object result = f.get(1000, TimeUnit.MILLISECONDS);
Object result = f.get(10000, TimeUnit.MILLISECONDS);
long elapsed = System.currentTimeMillis() - start;
assertTrue(elapsed >= 200);
assertNotNull(result);
@@ -109,7 +108,7 @@ public class AsyncGatewayTests {
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Future<Message<?>> f = service.returnMessage("foo");
try {
f.get(1000, TimeUnit.MILLISECONDS);
f.get(10000, TimeUnit.MILLISECONDS);
fail("Expected Exception");
}
catch (ExecutionException e) {
@@ -167,7 +166,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
CustomFuture f = service.returnCustomFuture("foo");
String result = f.get(1000, TimeUnit.MILLISECONDS);
String result = f.get(10000, TimeUnit.MILLISECONDS);
assertEquals("foobar", result);
assertEquals(Thread.currentThread(), f.thread);
}
@@ -188,7 +187,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
CustomFuture f = (CustomFuture) service.returnCustomFutureWithTypeFuture("foo");
String result = f.get(1000, TimeUnit.MILLISECONDS);
String result = f.get(10000, TimeUnit.MILLISECONDS);
assertEquals("foobar", result);
assertEquals(Thread.currentThread(), f.thread);
}
@@ -219,7 +218,7 @@ public class AsyncGatewayTests {
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Future<String> f = service.returnString("foo");
long start = System.currentTimeMillis();
Object result = f.get(1000, TimeUnit.MILLISECONDS);
Object result = f.get(10000, TimeUnit.MILLISECONDS);
long elapsed = System.currentTimeMillis() - start;
assertTrue(elapsed >= 200 - safety);
assertNotNull(result);
@@ -239,7 +238,7 @@ public class AsyncGatewayTests {
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Future<?> f = service.returnSomething("foo");
long start = System.currentTimeMillis();
Object result = f.get(1000, TimeUnit.MILLISECONDS);
Object result = f.get(10000, TimeUnit.MILLISECONDS);
long elapsed = System.currentTimeMillis() - start;
assertTrue(elapsed >= 200 - safety);
assertTrue(result instanceof String);
@@ -260,7 +259,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Promise<Message<?>> promise = service.returnMessagePromise("foo");
Object result = promise.await(1, TimeUnit.SECONDS);
Object result = promise.await(10, TimeUnit.SECONDS);
assertEquals("foobar", ((Message<?>) result).getPayload());
}
@@ -277,7 +276,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Promise<String> promise = service.returnStringPromise("foo");
Object result = promise.await(1, TimeUnit.SECONDS);
Object result = promise.await(10, TimeUnit.SECONDS);
assertEquals("foobar", result);
}
@@ -294,7 +293,7 @@ public class AsyncGatewayTests {
proxyFactory.afterPropertiesSet();
TestEchoService service = (TestEchoService) proxyFactory.getObject();
Promise<?> promise = service.returnSomethingPromise("foo");
Object result = promise.await(1, TimeUnit.SECONDS);
Object result = promise.await(10, TimeUnit.SECONDS);
assertNotNull(result);
assertEquals("foobar", result);
}
@@ -324,7 +323,7 @@ public class AsyncGatewayTests {
}
});
latch.await(1, TimeUnit.SECONDS);
latch.await(10, TimeUnit.SECONDS);
assertEquals("foobar", result.get());
}

14
spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml
View File

@@ -93,13 +93,25 @@
<si:channel id="usageChannelConcurrent">
<si:dispatcher task-executor="executor"/>
</si:channel>
<file:outbound-channel-adapter channel="usageChannelConcurrent"
filename-generator-expression="'fileToAppendConcurrent.txt'"
mode="APPEND"
directory="test"/>
directory="test">
<file:request-handler-advice-chain>
<bean class="org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice">
<property name="onSuccessExpression" value="@fileWriteLatch.countDown()"/>
</bean>
</file:request-handler-advice-chain>
</file:outbound-channel-adapter>
<bean id="customFileNameGenerator" class="org.springframework.integration.file.config.CustomFileNameGenerator"/>
<bean id="fileWriteLatch" class="java.util.concurrent.CountDownLatch">
<constructor-arg value="2"/>
</bean>
<context:property-placeholder/>
<task:executor id="executor" pool-size="100"/>

17
spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java
View File

@@ -25,6 +25,8 @@ import static org.junit.Assert.assertTrue;
import java.io.File;
import java.lang.reflect.Method;
import java.nio.charset.Charset;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Assert;
import org.junit.Test;
@@ -33,15 +35,15 @@ import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.expression.Expression;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.file.DefaultFileNameGenerator;
import org.springframework.integration.file.FileWritingMessageHandler;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.FileCopyUtils;
@@ -55,6 +57,7 @@ import org.springframework.util.ReflectionUtils;
* @author Oleg Zhurakousky
* @author Gunnar Hillert
* @author Tony Falabella
* @author Artem Bilan
*
*/
@ContextConfiguration
@@ -100,6 +103,9 @@ public class FileOutboundChannelAdapterParserTests {
@Autowired
MessageChannel usageChannelConcurrent;
@Autowired
CountDownLatch fileWriteLatch;
private volatile static int adviceCalled;
@Test
@@ -313,7 +319,8 @@ public class FileOutboundChannelAdapterParserTests {
usageChannelConcurrent.send(new GenericMessage<String>(bString));
}
Thread.sleep(2000);
assertTrue(this.fileWriteLatch.await(10, TimeUnit.SECONDS));
String actualFileContent = new String(FileCopyUtils.copyToByteArray(testFile));
int beginningIndex = 0;
for (int i = 0; i < 2; i++) {

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java
View File

@@ -130,7 +130,9 @@ public class ConnectionToConnectionTests {
int serverCloses = 0;
int clientExceptions = 0;
Message<TcpConnectionEvent> eventMessage;
while ((eventMessage = (Message<TcpConnectionEvent>) events.receive(1000)) != null) {
int i = 0;
while (i++ < (expectExceptionOnClose ? 600 : 400)
&& (eventMessage = (Message<TcpConnectionEvent>) events.receive(10000)) != null) {
TcpConnectionEvent event = eventMessage.getPayload();
if (event.getConnectionFactoryName().startsWith("client")) {
if (event instanceof TcpConnectionOpenEvent) {

11
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

@@ -52,6 +52,7 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.net.ServerSocketFactory;
import org.junit.ClassRule;
import org.junit.Test;
import org.mockito.Mockito;
@@ -71,6 +72,7 @@ import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionF
import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.support.LogAdjustingTestSupport;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
@@ -83,6 +85,9 @@ import org.springframework.messaging.support.GenericMessage;
*/
public class TcpOutboundGatewayTests extends LogAdjustingTestSupport {
@ClassRule
public static LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@Test
public void testGoodNetSingle() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
@@ -367,7 +372,7 @@ public class TcpOutboundGatewayTests extends LogAdjustingTestSupport {
logger.debug("Read " + request);
ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
if (i < 2) {
Thread.sleep(1000);
Thread.sleep(2000);
}
oos.writeObject(request.replace("Test", "Reply"));
logger.debug("Replied to " + request);
@@ -400,7 +405,7 @@ public class TcpOutboundGatewayTests extends LogAdjustingTestSupport {
Expression remoteTimeoutExpression = Mockito.mock(Expression.class);
when(remoteTimeoutExpression.getValue(Mockito.any(EvaluationContext.class), Mockito.any(Message.class),
Mockito.eq(Long.class))).thenReturn(500L, 10000L);
Mockito.eq(Long.class))).thenReturn(50L, 10000L);
gateway.setRemoteTimeoutExpression(remoteTimeoutExpression);
@@ -421,7 +426,7 @@ public class TcpOutboundGatewayTests extends LogAdjustingTestSupport {
}
// wait until the server side has processed both requests
assertTrue(serverLatch.await(10, TimeUnit.SECONDS));
assertTrue(serverLatch.await(30, TimeUnit.SECONDS));
List<String> replies = new ArrayList<String>();
int timeouts = 0;
for (int i = 0; i < 2; i++) {

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java
View File

@@ -448,13 +448,13 @@ public class CachingClientConnectionFactoryTests {
public void integrationTest() throws Exception {
TestingUtilities.waitListening(serverCf, null);
outbound.send(new GenericMessage<String>("Hello, world!"));
Message<?> m = inbound.receive(1000);
Message<?> m = inbound.receive(10000);
assertNotNull(m);
String connectionId = m.getHeaders().get(IpHeaders.CONNECTION_ID, String.class);
// assert we use the same connection from the pool
outbound.send(new GenericMessage<String>("Hello, world!"));
m = inbound.receive(1000);
m = inbound.receive(10000);
assertNotNull(m);
assertEquals(connectionId, m.getHeaders().get(IpHeaders.CONNECTION_ID, String.class));
}

2
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml
View File

@@ -81,7 +81,7 @@
</bean>
<int:outbound-channel-adapter channel="second" expression="@successfulLatch.countDown()">
<int:poller fixed-delay="1000">
<int:poller fixed-delay="100">
<int:transactional transaction-manager="transactionManager"/>
</int:poller>
</int:outbound-channel-adapter>

4
spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java
View File

@@ -405,7 +405,7 @@ public class OutboundGatewayFunctionTests {
gateway.setUseReplyContainer(true);
gateway.setIdleReplyContainerTimeout(1, TimeUnit.SECONDS);
gateway.setRequiresReply(true);
gateway.setReceiveTimeout(10000);
gateway.setReceiveTimeout(20000);
gateway.afterPropertiesSet();
gateway.start();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@@ -413,7 +413,7 @@ public class OutboundGatewayFunctionTests {
public void run() {
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getTemplateConnectionFactory());
template.setReceiveTimeout(10000);
template.setReceiveTimeout(20000);
receiveAndSend(template);
receiveAndSend(template);
}

12
spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java
View File

@@ -54,6 +54,8 @@ import javax.mail.search.FlagTerm;
import javax.mail.search.FromTerm;
import javax.mail.search.SearchTerm;
import com.sun.mail.imap.IMAPFolder;
import com.sun.mail.imap.IMAPMessage;
import org.apache.commons.logging.Log;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -87,9 +89,6 @@ import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.FileCopyUtils;
import com.sun.mail.imap.IMAPFolder;
import com.sun.mail.imap.IMAPMessage;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
@@ -129,8 +128,7 @@ public class ImapMailReceiverTests {
public SearchTerm generateSearchTerm(Flags supportedFlags, Folder folder) {
try {
FromTerm fromTerm = new FromTerm(new InternetAddress("bar@baz"));
AndTerm andTerm = new AndTerm(fromTerm, new FlagTerm(new Flags(Flags.Flag.SEEN), false));
return andTerm;
return new AndTerm(fromTerm, new FlagTerm(new Flags(Flag.SEEN), false));
}
catch (AddressException e) {
throw new RuntimeException(e);
@@ -171,11 +169,11 @@ public class ImapMailReceiverTests {
adapter.start();
@SuppressWarnings("unchecked")
org.springframework.messaging.Message<MimeMessage> received =
(org.springframework.messaging.Message<MimeMessage>) channel.receive(6000);
(org.springframework.messaging.Message<MimeMessage>) channel.receive(10000);
assertNotNull(received);
assertNotNull(received.getPayload().getReceivedDate());
assertTrue(received.getPayload().getLineCount() > -1);
assertNotNull(channel.receive(6000)); // new message after idle
assertNotNull(channel.receive(10000)); // new message after idle
assertNull(channel.receive(10000)); // no new message after second and third idle
verify(logger).debug("Canceling IDLE");
taskScheduler.shutdown();

7
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
View File

@@ -237,9 +237,6 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
}
if (this.client.isConnected()) {
this.connected = true;
if (this.reconnectFuture != null) {
cancelReconnect();
}
String message = "Connected and subscribed to " + Arrays.asList(getTopic());
if (logger.isDebugEnabled()) {
logger.debug(message);
@@ -247,6 +244,10 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(new MqttSubscribedEvent(this, message));
}
// cancel() after the publish in case we are on that thread; a send to a QueueChannel would fail.
if (this.reconnectFuture != null) {
cancelReconnect();
}
}
}

14
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpServerTests.java
View File

@@ -30,6 +30,7 @@ import java.security.PublicKey;
import java.security.spec.RSAPublicKeySpec;
import java.util.Collections;
import com.jcraft.jsch.ChannelSftp.LsEntry;
import org.apache.sshd.SshServer;
import org.apache.sshd.common.NamedFactory;
import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory;
@@ -45,11 +46,8 @@ import org.junit.Test;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.core.io.ClassPathResource;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.test.util.SocketUtils;
import org.springframework.util.StreamUtils;
import com.jcraft.jsch.ChannelSftp.LsEntry;
/**
* *
* @author Gary Russell
@@ -61,7 +59,6 @@ public class SftpServerTests {
@Test
public void testUcPw() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
SshServer server = SshServer.setUpDefaultServer();
try {
server.setPasswordAuthenticator(new PasswordAuthenticator() {
@@ -71,7 +68,7 @@ public class SftpServerTests {
return true;
}
});
server.setPort(port);
server.setPort(0);
server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser"));
server.setSubsystemFactories(Collections.<NamedFactory<Command>>singletonList(new SftpSubsystem.Factory()));
final String pathname = System.getProperty("java.io.tmpdir") + File.separator + "sftptest" + File.separator;
@@ -81,7 +78,7 @@ public class SftpServerTests {
DefaultSftpSessionFactory f = new DefaultSftpSessionFactory();
f.setHost("localhost");
f.setPort(port);
f.setPort(server.getPort());
f.setUser("user");
f.setPassword("pass");
f.setAllowUnknownKeys(true);
@@ -105,7 +102,6 @@ public class SftpServerTests {
private void testKeyExchange(String pubKey, String privKey, String passphrase)
throws Exception, IOException, InterruptedException {
final int port = SocketUtils.findAvailableServerSocket();
SshServer server = SshServer.setUpDefaultServer();
final PublicKey allowedKey = decodePublicKey(pubKey);
try {
@@ -117,7 +113,7 @@ public class SftpServerTests {
}
});
server.setPort(port);
server.setPort(0);
server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser"));
server.setSubsystemFactories(Collections.<NamedFactory<Command>>singletonList(new SftpSubsystem.Factory()));
final String pathname = System.getProperty("java.io.tmpdir") + File.separator + "sftptest" + File.separator;
@@ -127,7 +123,7 @@ public class SftpServerTests {
DefaultSftpSessionFactory f = new DefaultSftpSessionFactory();
f.setHost("localhost");
f.setPort(port);
f.setPort(server.getPort());
f.setUser("user");
f.setAllowUnknownKeys(true);
InputStream stream = new ClassPathResource(privKey).getInputStream();

15
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.stomp;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ScheduledFuture;
@@ -190,26 +191,27 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
}
private void scheduleReconnect(Throwable e) {
if (this.reconnectFuture != null) {
this.reconnectFuture.cancel(true);
this.reconnectFuture = null;
}
this.connecting = this.connected = false;
logger.error("STOMP connect error.", e);
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(
new StompConnectionFailedEvent(this, e));
}
// cancel() after the publish in case we are on that thread; a send to a QueueChannel would fail.
if (this.reconnectFuture != null) {
this.reconnectFuture.cancel(true);
this.reconnectFuture = null;
}
this.reconnectFuture = this.stompClient.getTaskScheduler()
.scheduleWithFixedDelay(new Runnable() {
.schedule(new Runnable() {
@Override
public void run() {
connect();
}
}, this.recoveryInterval);
}, new Date(System.currentTimeMillis() + this.recoveryInterval));
}
@Override
@@ -349,6 +351,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
@Override
public void handleTransportError(StompSession session, Throwable exception) {
logger.error("STOMP transport error for session: [" + session + "]", exception);
if (exception instanceof ConnectionLostException) {
this.session = null;
scheduleReconnect(exception);

6
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java
View File

@@ -39,7 +39,6 @@ import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.simp.stomp.ConnectionLostException;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompFrameHandler;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
@@ -304,10 +303,7 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement
@Override
public void handleTransportError(StompSession session, Throwable exception) {
if (exception instanceof ConnectionLostException) {
StompInboundChannelAdapter.this.stompSession = null;
}
logger.error("STOMP transport error for session: [" + session + "]", exception);
StompInboundChannelAdapter.this.stompSession = null;
}
}

7
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java
View File

@@ -261,12 +261,7 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli
@Override
public synchronized void handleTransportError(StompSession session, Throwable exception) {
StompMessageHandler.this.transportError = exception;
if (exception instanceof ConnectionLostException) {
StompMessageHandler.this.stompSession = null;
}
else {
logger.error("STOMP transport error for session: [" + session + "]", exception);
}
StompMessageHandler.this.stompSession = null;
}
}

12
spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java
View File

@@ -51,6 +51,7 @@ import org.springframework.integration.stomp.event.StompIntegrationEvent;
import org.springframework.integration.stomp.event.StompReceiptEvent;
import org.springframework.integration.stomp.event.StompSessionConnectedEvent;
import org.springframework.integration.test.rule.Log4jLevelAdjuster;
import org.springframework.integration.test.support.LogAdjustingTestSupport;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.websocket.TomcatWebSocketTestServer;
import org.springframework.messaging.Message;
@@ -92,10 +93,7 @@ import org.springframework.web.socket.sockjs.client.WebSocketTransport;
@ContextConfiguration(classes = StompInboundChannelAdapterWebSocketIntegrationTests.ContextConfiguration.class)
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class StompInboundChannelAdapterWebSocketIntegrationTests {
@Rule
public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE, "org.springframework");
public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdjustingTestSupport {
@Value("#{server.serverContext}")
private ConfigurableApplicationContext serverContext;
@@ -115,6 +113,10 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests {
@Autowired
private StompInboundChannelAdapter stompInboundChannelAdapter;
public StompInboundChannelAdapterWebSocketIntegrationTests() {
super("org.springframework", "org.springframework.integration.stomp");
}
@Test
public void testWebSocketStompClient() throws Exception {
Message<?> eventMessage = this.stompEvents.receive(10000);
@@ -240,7 +242,7 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests {
}
@Bean
public WebSocketStompClient stompClient(@Qualifier("taskScheduler") TaskScheduler taskScheduler) {
public WebSocketStompClient stompClient(TaskScheduler taskScheduler) {
WebSocketStompClient webSocketStompClient = new WebSocketStompClient(webSocketClient());
webSocketStompClient.setMessageConverter(new MappingJackson2MessageConverter());
webSocketStompClient.setTaskScheduler(taskScheduler);

7
spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java
View File

@@ -55,6 +55,7 @@ import org.springframework.integration.stomp.event.StompExceptionEvent;
import org.springframework.integration.stomp.event.StompIntegrationEvent;
import org.springframework.integration.stomp.event.StompReceiptEvent;
import org.springframework.integration.stomp.event.StompSessionConnectedEvent;
import org.springframework.integration.test.support.LogAdjustingTestSupport;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.websocket.TomcatWebSocketTestServer;
@@ -96,7 +97,7 @@ import org.springframework.web.socket.sockjs.client.WebSocketTransport;
@ContextConfiguration(classes = StompMessageHandlerWebSocketIntegrationTests.ContextConfiguration.class)
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class StompMessageHandlerWebSocketIntegrationTests {
public class StompMessageHandlerWebSocketIntegrationTests extends LogAdjustingTestSupport {
@ClassRule
public static LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@@ -112,6 +113,10 @@ public class StompMessageHandlerWebSocketIntegrationTests {
@Qualifier("stompEvents")
private PollableChannel stompEvents;
public StompMessageHandlerWebSocketIntegrationTests() {
super("org.springframework", "org.springframework.integration.stomp");
}
@Test
public void testStompMessageHandler() throws InterruptedException {
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.SEND);

14
spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 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.
@@ -57,7 +57,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine
private volatile CountDownLatch connectionLatch;
private WebSocketSession clientSession;
private volatile WebSocketSession clientSession;
private volatile Throwable openConnectionException;
@@ -106,7 +106,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine
*/
@Override
public WebSocketSession getSession(String sessionId) {
if (this.isRunning()) {
if (isRunning()) {
try {
this.connectionLatch.await(this.connectionTimeout, TimeUnit.SECONDS);
}
@@ -146,9 +146,11 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine
}
@Override
public void start() {
this.connectionLatch = new CountDownLatch(1);
this.connectionManager.start();
public synchronized void start() {
if (!isRunning()) {
this.connectionLatch = new CountDownLatch(1);
this.connectionManager.start();
}
}
@Override

32
spring-integration-websocket/src/test/java/org/springframework/integration/websocket/TomcatWebSocketTestServer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 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.
@@ -20,14 +20,11 @@ import java.io.File;
import java.io.IOException;
import org.apache.catalina.Context;
import org.apache.catalina.connector.Connector;
import org.apache.catalina.startup.Tomcat;
import org.apache.coyote.http11.Http11NioProtocol;
import org.apache.tomcat.websocket.server.WsContextListener;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.SocketUtils;
import org.springframework.web.context.support.AnnotationConfigWebApplicationContext;
import org.springframework.web.servlet.DispatcherServlet;
@@ -40,40 +37,29 @@ public class TomcatWebSocketTestServer implements InitializingBean, DisposableBe
private final Tomcat tomcatServer;
private final int port;
private final AnnotationConfigWebApplicationContext serverContext;
public TomcatWebSocketTestServer(Class<?>... serverConfigs) {
this.port = SocketUtils.findAvailableTcpPort();
Connector connector = new Connector(Http11NioProtocol.class.getName());
connector.setPort(this.port);
File baseDir = createTempDir("tomcat");
String baseDirPath = baseDir.getAbsolutePath();
this.tomcatServer = new Tomcat();
this.tomcatServer.setBaseDir(baseDirPath);
this.tomcatServer.setPort(this.port);
this.tomcatServer.getService().addConnector(connector);
this.tomcatServer.setConnector(connector);
this.tomcatServer.setPort(0);
this.tomcatServer.setBaseDir(createTempDir());
this.serverContext = new AnnotationConfigWebApplicationContext();
this.serverContext.register(serverConfigs);
Context context = this.tomcatServer.addContext("", System.getProperty("java.io.tmpdir"));
context.addApplicationListener(WsContextListener.class.getName());
Tomcat.addServlet(context, "dispatcherServlet", new DispatcherServlet(this.serverContext)).setAsyncSupported(true);
Tomcat.addServlet(context, "dispatcherServlet", new DispatcherServlet(this.serverContext))
.setAsyncSupported(true);
context.addServletMapping("/", "dispatcherServlet");
}
private File createTempDir(String prefix) {
private String createTempDir() {
try {
File tempFolder = File.createTempFile(prefix + ".", "." + this.port);
File tempFolder = File.createTempFile("tomcat.", ".workDir");
tempFolder.delete();
tempFolder.mkdir();
tempFolder.deleteOnExit();
return tempFolder;
return tempFolder.getAbsolutePath();
}
catch (IOException ex) {
throw new RuntimeException("Unable to create temp directory", ex);
@@ -85,7 +71,7 @@ public class TomcatWebSocketTestServer implements InitializingBean, DisposableBe
}
public String getWsBaseUrl() {
return "ws://localhost:" + this.port;
return "ws://localhost:" + this.tomcatServer.getConnector().getLocalPort();
}
@Override