From 9d8e2b61f6125b6ccc6dcb0d817ca828abb2b476 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 13 Nov 2015 22:39:00 -0500 Subject: [PATCH] Fixes for Sporadic Test Failures * `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 --- .../AggregatorIntegrationTests.java | 10 +++--- .../channel/PriorityChannelTests.java | 26 ++++++++------- .../config/xml/GatewayParserTests.java | 5 ++- .../configuration/EnableIntegrationTests.java | 4 +++ .../core/AsyncMessagingTemplateTests.java | 25 +++++++++------ .../gateway/AsyncGatewayTests.java | 27 ++++++++-------- ...boundChannelAdapterParserTests-context.xml | 14 +++++++- ...FileOutboundChannelAdapterParserTests.java | 17 +++++++--- .../ip/tcp/ConnectionToConnectionTests.java | 4 ++- .../ip/tcp/TcpOutboundGatewayTests.java | 11 +++++-- .../CachingClientConnectionFactoryTests.java | 4 +-- .../TxTimeoutMessageStoreTests-context.xml | 2 +- .../jms/OutboundGatewayFunctionTests.java | 4 +-- .../mail/ImapMailReceiverTests.java | 12 +++---- .../MqttPahoMessageDrivenChannelAdapter.java | 7 ++-- .../sftp/session/SftpServerTests.java | 14 +++----- .../stomp/AbstractStompSessionManager.java | 15 +++++---- .../inbound/StompInboundChannelAdapter.java | 6 +--- .../stomp/outbound/StompMessageHandler.java | 7 +--- ...annelAdapterWebSocketIntegrationTests.java | 12 ++++--- ...ssageHandlerWebSocketIntegrationTests.java | 7 +++- .../websocket/ClientWebSocketContainer.java | 14 ++++---- .../websocket/TomcatWebSocketTestServer.java | 32 ++++++------------- 23 files changed, 150 insertions(+), 129 deletions(-) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java index 437eec7a6a..b27d3d0154 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java @@ -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(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(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(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)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java index 9437ee3183..6e4db52489 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java @@ -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("test-1")); executor.execute(new Runnable() { + @Override public void run() { sentSecondMessage.set(channel.send(new GenericMessage("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("test-1")); executor.execute(new Runnable() { @Override public void run() { sentSecondMessage.set(channel.send(new GenericMessage("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> { + @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); } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java index 94779d637b..346f8fecf4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java @@ -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 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")); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java index b1aa3456e6..38ed0bee58 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java @@ -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")); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java index 19a2000fb4..f214d1d144 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java @@ -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> result = template.asyncReceive(); sendMessageAfterDelay(channel, new GenericMessage("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> result = template.asyncReceive(channel); sendMessageAfterDelay(channel, new GenericMessage("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 result = template.asyncConvertSendAndReceive(new Integer(123), new TestMessagePostProcessor()); + Future 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 message, final int delay) { + private static void sendMessageAfterDelay(final MessageChannel channel, final GenericMessage 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(); } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index 552df90c63..98ea8054ea 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -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> 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> 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 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> 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 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()); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml index 3d8a4b11a4..72d130e294 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml @@ -93,13 +93,25 @@ + + + directory="test"> + + + + + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java index 49af763b83..1893106dc2 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java @@ -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(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++) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java index 493b10f58c..004842b42d 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java @@ -130,7 +130,9 @@ public class ConnectionToConnectionTests { int serverCloses = 0; int clientExceptions = 0; Message eventMessage; - while ((eventMessage = (Message) events.receive(1000)) != null) { + int i = 0; + while (i++ < (expectExceptionOnClose ? 600 : 400) + && (eventMessage = (Message) events.receive(10000)) != null) { TcpConnectionEvent event = eventMessage.getPayload(); if (event.getConnectionFactoryName().startsWith("client")) { if (event instanceof TcpConnectionOpenEvent) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index 8af58f98c1..3d2a84fa39 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -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 replies = new ArrayList(); int timeouts = 0; for (int i = 0; i < 2; i++) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index 6c791b880c..484377a947 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java @@ -448,13 +448,13 @@ public class CachingClientConnectionFactoryTests { public void integrationTest() throws Exception { TestingUtilities.waitListening(serverCf, null); outbound.send(new GenericMessage("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("Hello, world!")); - m = inbound.receive(1000); + m = inbound.receive(10000); assertNotNull(m); assertEquals(connectionId, m.getHeaders().get(IpHeaders.CONNECTION_ID, String.class)); } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml index 7d82ee2ab2..260161dbbd 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/TxTimeoutMessageStoreTests-context.xml @@ -81,7 +81,7 @@ - + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java index d7738f7eef..8406fdff2d 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java @@ -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); } diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java index c9be6efc77..0c298fa4a5 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java @@ -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 received = - (org.springframework.messaging.Message) channel.receive(6000); + (org.springframework.messaging.Message) 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(); diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java index ab25f4201f..11b3bf9e9b 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java @@ -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(); + } } } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpServerTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpServerTests.java index c2830ae65c..dd62be4f09 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpServerTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpServerTests.java @@ -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.>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.>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(); diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java index f1050b2220..0b3e85f213 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java @@ -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); diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java index 8a92b0c60c..d07f768d4a 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java @@ -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; } } diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java index ec0871d730..38e737fc2f 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java @@ -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; } } diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java index 326f0e5314..da35e7b1de 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java @@ -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); diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java index 1ac1cf2161..ff36b1ce79 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java @@ -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); diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java index bdf76ff148..5fea26c82f 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java @@ -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 diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/TomcatWebSocketTestServer.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/TomcatWebSocketTestServer.java index 43ff1099ac..d2dabc7b70 100644 --- a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/TomcatWebSocketTestServer.java +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/TomcatWebSocketTestServer.java @@ -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