diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java index 91afbda1b9..e147e487b5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java @@ -308,11 +308,14 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS Iterator messageIds = messageGroupMetadata.messageIdIterator(); while (messageIds.hasNext()){ - if (raw){ - messages.add(getRawMessage(messageIds.next())); - } - else { - messages.add(getMessage(messageIds.next())); + UUID next = messageIds.next(); + if (next != null) { + if (raw){ + messages.add(getRawMessage(next)); + } + else { + messages.add(getMessage(next)); + } } } 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 4148885b79..19a2000fb4 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 @@ -150,7 +150,7 @@ public class AsyncMessagingTemplateTests { Future> result = template.asyncReceive(); 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); @@ -180,7 +180,7 @@ public class AsyncMessagingTemplateTests { Future> result = template.asyncReceive("testChannel"); 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; assertTrue(elapsed >= 200-safety); @@ -202,7 +202,7 @@ public class AsyncMessagingTemplateTests { Future result = template.asyncReceiveAndConvert(); 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()); @@ -216,7 +216,7 @@ public class AsyncMessagingTemplateTests { Future result = template.asyncReceiveAndConvert(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()); @@ -234,7 +234,7 @@ public class AsyncMessagingTemplateTests { Future result = template.asyncReceiveAndConvert("testChannel"); 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; assertTrue(elapsed >= 200-safety); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java index 6903cfaf9a..78771aec91 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java @@ -41,6 +41,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Iwein Fuld * @author Gary Russell + * @author Artem Bilan */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration @@ -87,7 +88,7 @@ public class FileReadingMessageSourceIntegrationTests { } @After - public void cleanoutInputDir() throws Exception { + public void cleanupInputDir() throws Exception { File[] listFiles = inputDir.listFiles(); for (int i = 0; i < listFiles.length; i++) { listFiles[i].delete(); @@ -140,11 +141,12 @@ public class FileReadingMessageSourceIntegrationTests { assertNull(pollableFileSource.receive()); } - @Test(timeout = 6000) + @Test @Repeat(5) public void concurrentProcessing() throws Exception { CountDownLatch go = new CountDownLatch(1); - Runnable succesfulConsumer = new Runnable() { + Runnable successfulConsumer = new Runnable() { + @Override public void run() { Message received = pollableFileSource.receive(); @@ -154,8 +156,10 @@ public class FileReadingMessageSourceIntegrationTests { } pollableFileSource.onSend(received); } + }; Runnable failingConsumer = new Runnable() { + @Override public void run() { Message received = pollableFileSource.receive(); @@ -163,12 +167,13 @@ public class FileReadingMessageSourceIntegrationTests { pollableFileSource.onFailure(received); } } + }; - CountDownLatch succesfulDone = doConcurrently(3, succesfulConsumer, go); + CountDownLatch successfulDone = doConcurrently(3, successfulConsumer, go); CountDownLatch failingDone = doConcurrently(10, failingConsumer, go); go.countDown(); try { - succesfulDone.await(); + successfulDone.await(); failingDone.await(); } catch (InterruptedException e) { @@ -187,7 +192,8 @@ public class FileReadingMessageSourceIntegrationTests { * * @param numberOfThreads how many threads to spawn * @param runnable the runnable that should be run by all the threads - * @param start the {@link java.util.concurrent.CountDownLatch} instance telling it when to assume everything works + * @param start the {@link java.util.concurrent.CountDownLatch} instance + * telling it when to assume everything works * @return a latch that will be counted down once all threads have run their * runnable. */ @@ -210,8 +216,10 @@ public class FileReadingMessageSourceIntegrationTests { runnable.run(); done.countDown(); } + }).start(); } return done; } + } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests-context.xml index b0fe39ab9e..b3a7c65be1 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests-context.xml @@ -1,14 +1,10 @@ + http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd"> - - - - - + diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CqInboundChannelAdapterTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CqInboundChannelAdapterTests.java index 222aa8152b..a1c9c57454 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CqInboundChannelAdapterTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CqInboundChannelAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-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. You may obtain a copy of the License at @@ -10,6 +10,7 @@ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * specific language governing permissions and limitations under the License. */ + package org.springframework.integration.gemfire.inbound; import static org.junit.Assert.assertEquals; @@ -78,7 +79,7 @@ public class CqInboundChannelAdapterTests { @Test public void testCqEvent() throws InterruptedException { region.put("one",1); - Message msg = outputChannel1.receive(1000); + Message msg = outputChannel1.receive(10000); assertNotNull(msg); assertTrue(msg.getPayload() instanceof CqEvent); } @@ -86,7 +87,7 @@ public class CqInboundChannelAdapterTests { @Test public void testPayloadExpression() throws InterruptedException { region.put("one",1); - Message msg = outputChannel2.receive(1000); + Message msg = outputChannel2.receive(10000); assertNotNull(msg); assertEquals(1,msg.getPayload()); } @@ -108,4 +109,5 @@ public class CqInboundChannelAdapterTests { throw new IllegalStateException("Cannot communicate with forked VM", ex); } } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index 69bf863cb8..05b8f35c5d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -187,7 +187,9 @@ public abstract class AbstractServerConnectionFactory } protected void publishServerExceptionEvent(Exception e) { - getApplicationEventPublisher().publishEvent(new TcpConnectionServerExceptionEvent(this, e)); + if (getApplicationEventPublisher() != null) { + getApplicationEventPublisher().publishEvent(new TcpConnectionServerExceptionEvent(this, e)); + } } } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java index 0c1c649bba..4d94a0b62a 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-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,8 +57,6 @@ import org.springframework.transaction.support.TransactionTemplate; // @Transactional public class JdbcPollingChannelAdapterParserTests { - final long receiveTimeout = 5000; - private JdbcTemplate jdbcTemplate; private MessagingTemplate messagingTemplate; @@ -202,7 +200,7 @@ public class JdbcPollingChannelAdapterParserTests { PollableChannel pollableChannel = this.appCtx.getBean("target", PollableChannel.class); this.messagingTemplate = new MessagingTemplate(); this.messagingTemplate.setDefaultDestination(pollableChannel); - this.messagingTemplate.setReceiveTimeout(500); + this.messagingTemplate.setReceiveTimeout(10000); } protected void setupJdbcTemplate() { diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java index 43a7ac927c..340c53472c 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-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. You may obtain a copy of the License at @@ -10,6 +10,7 @@ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * specific language governing permissions and limitations under the License. */ + package org.springframework.integration.jdbc.store.channel; import static org.junit.Assert.assertEquals; @@ -191,7 +192,7 @@ public abstract class AbstractTxTimeoutMessageStoreTests { } executorService.shutdown(); - assertTrue(executorService.awaitTermination(5, TimeUnit.SECONDS)); + assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS)); } @Test @@ -200,7 +201,7 @@ public abstract class AbstractTxTimeoutMessageStoreTests { this.first.send(new GenericMessage("test")); } - assertTrue(this.successfulLatch.await(5, TimeUnit.SECONDS)); + assertTrue(this.successfulLatch.await(20, TimeUnit.SECONDS)); assertEquals(0, errorAtomicInteger.get()); } @@ -247,31 +248,31 @@ public abstract class AbstractTxTimeoutMessageStoreTests { message = MessageBuilder.withPayload("31").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 3).build(); priorityChannel.send(message); - Message receive = priorityChannel.receive(1000); + Message receive = priorityChannel.receive(10000); assertNotNull(receive); assertEquals("3", receive.getPayload()); - receive = priorityChannel.receive(1000); + receive = priorityChannel.receive(10000); assertNotNull(receive); assertEquals("31", receive.getPayload()); - receive = priorityChannel.receive(1000); + receive = priorityChannel.receive(10000); assertNotNull(receive); assertEquals("2", receive.getPayload()); - receive = priorityChannel.receive(1000); + receive = priorityChannel.receive(10000); assertNotNull(receive); assertEquals("1", receive.getPayload()); - receive = priorityChannel.receive(1000); + receive = priorityChannel.receive(10000); assertNotNull(receive); assertEquals("0", receive.getPayload()); - receive = priorityChannel.receive(1000); + receive = priorityChannel.receive(10000); assertNotNull(receive); assertEquals("-1", receive.getPayload()); - receive = priorityChannel.receive(1000); + receive = priorityChannel.receive(10000); assertNotNull(receive); assertEquals("none", receive.getPayload()); } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 58191bc1a3..17d18c72ce 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -1200,7 +1200,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp private class GatewayReplyListenerContainer extends DefaultMessageListenerContainer { - private Destination replyDestination; + private volatile Destination replyDestination; @Override protected Destination resolveDestinationName(Session session, String destinationName) throws JMSException { diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java index d893ac43fb..81e444a1ea 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java @@ -62,7 +62,7 @@ public class MBeanAttributeFilterTests { @Autowired private String domain; - private final long testTimeout = 1000L; + private final long testTimeout = 10000L; @Test public void testAttributeFilter() { diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests-context.xml index b2bb2d3f14..1ed5b18c5d 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests-context.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests-context.xml @@ -1,9 +1,8 @@ + http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> @@ -23,8 +22,12 @@ - + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java index cd1c04c4a6..b09f51ddb0 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java @@ -10,6 +10,7 @@ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * specific language governing permissions and limitations under the License. */ + package org.springframework.integration.monitor; import static org.junit.Assert.assertEquals; @@ -33,6 +34,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Dave Syer * @author Gary Russell + * @author Artem Bilan * @since 2.0 */ @ContextConfiguration @@ -46,6 +48,9 @@ public class ChannelIntegrationTests { @Autowired private PollableChannel intermediate; + @Autowired + private PollableChannel sourceChannel; + @Autowired private IntegrationMBeanExporter messageChannelsMonitor; @@ -85,7 +90,7 @@ public class ChannelIntegrationTests { assertEquals(3, handlerMetrics.getHandleCount()); assertEquals(1, handlerMetrics.getErrorCount()); - Thread.sleep(50); + assertNotNull(this.sourceChannel.receive(10000)); assertTrue(messageChannelsMonitor.getSourceMessageCount("source") > 0); assertTrue(messageChannelsMonitor.getSourceMetrics("source").getMessageCount() > 0); diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MonitorTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MonitorTests-context.xml index 0c3700c7e4..577c7ac06a 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MonitorTests-context.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MonitorTests-context.xml @@ -16,9 +16,11 @@ - + + + - + diff --git a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/BaseJpaPollingChannelAdapterTests-context.xml b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/BaseJpaPollingChannelAdapterTests-context.xml index a38a34e83b..54e74ac027 100644 --- a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/BaseJpaPollingChannelAdapterTests-context.xml +++ b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/BaseJpaPollingChannelAdapterTests-context.xml @@ -51,7 +51,7 @@ - + diff --git a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapterTests.java b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapterTests.java index eb85bd339e..5cea995228 100644 --- a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapterTests.java +++ b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapterTests.java @@ -39,7 +39,7 @@ import org.springframework.integration.jpa.core.JpaExecutor; import org.springframework.integration.jpa.core.JpaOperations; import org.springframework.integration.jpa.test.Consumer; import org.springframework.integration.jpa.test.JpaTestUtils; -import org.springframework.integration.jpa.test.TestTrigger; +import org.springframework.integration.test.util.OnlyOnceTrigger; import org.springframework.integration.jpa.test.entity.StudentDomain; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.messaging.Message; @@ -83,7 +83,7 @@ public class JpaPollingChannelAdapterTests { private MessageChannel outputChannel; @Autowired - TestTrigger testTrigger; + OnlyOnceTrigger testTrigger; /** * In this test, a Jpa Polling Channel Adapter will use a plain entity class @@ -113,7 +113,7 @@ public class JpaPollingChannelAdapterTests { final Consumer consumer = new Consumer(); - received.add(consumer.poll(5000)); + received.add(consumer.poll(10000)); Message> message = received.get(0); @@ -156,7 +156,7 @@ public class JpaPollingChannelAdapterTests { final Consumer consumer = new Consumer(); - received.add(consumer.poll(5000)); + received.add(consumer.poll(10000)); Message> message = received.get(0); @@ -200,7 +200,7 @@ public class JpaPollingChannelAdapterTests { final Consumer consumer = new Consumer(); - received.add(consumer.poll(5000)); + received.add(consumer.poll(10000)); Message> message = received.get(0); @@ -242,7 +242,7 @@ public class JpaPollingChannelAdapterTests { final Consumer consumer = new Consumer(); - received.add(consumer.poll(5000)); + received.add(consumer.poll(10000)); Message> message = received.get(0); @@ -262,7 +262,7 @@ public class JpaPollingChannelAdapterTests { /** * In this test, a Jpa Polling Channel Adapter will use JpQL query - * to retrieve a list of records from the database. Additionaly, the records + * to retrieve a list of records from the database. Additionally, the records * will be deleted after the polling. * * @throws Exception @@ -292,7 +292,7 @@ public class JpaPollingChannelAdapterTests { final Consumer consumer = new Consumer(); - received.add(consumer.poll(5000)); + received.add(consumer.poll(10000)); Message> message = received.get(0); @@ -305,8 +305,10 @@ public class JpaPollingChannelAdapterTests { assertTrue(students.size() == 3); - assertEquals(Long.valueOf(0), entityManager.createQuery("select count(*) from Student", Long.class).getSingleResult()); + Long studentCount = waitForDeletes(students); + + assertEquals(Long.valueOf(0), studentCount); } private Long waitForDeletes(Collection students) throws InterruptedException { @@ -348,7 +350,7 @@ public class JpaPollingChannelAdapterTests { final Consumer consumer = new Consumer(); final List>> received = new ArrayList>>(); - received.add(consumer.poll(5000)); + received.add(consumer.poll(10000)); final Message> message = received.get(0); @@ -388,7 +390,7 @@ public class JpaPollingChannelAdapterTests { final Consumer consumer = new Consumer(); final List>> received = new ArrayList>>(); - received.add(consumer.poll(5000)); + received.add(consumer.poll(10000)); final Message> message = received.get(0); @@ -435,7 +437,7 @@ public class JpaPollingChannelAdapterTests { final Consumer consumer = new Consumer(); - received.add(consumer.poll(5000)); + received.add(consumer.poll(10000)); Message> message = received.get(0); @@ -478,7 +480,7 @@ public class JpaPollingChannelAdapterTests { final Consumer consumer = new Consumer(); - received.add(consumer.poll(5000)); + received.add(consumer.poll(10000)); Message> message = received.get(0); diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java index b32e2ddad2..3614b2d896 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java @@ -23,8 +23,6 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; -import java.util.concurrent.Executors; - import org.apache.activemq.broker.BrokerService; import org.junit.AfterClass; import org.junit.BeforeClass; @@ -55,8 +53,6 @@ import org.springframework.messaging.PollableChannel; import org.springframework.messaging.simp.stomp.Reactor2TcpStompClient; import org.springframework.messaging.simp.stomp.StompCommand; import org.springframework.messaging.support.GenericMessage; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.SocketUtils; @@ -81,11 +77,13 @@ public class StompServerIntegrationTests { activeMQBroker.getSystemUsage().getMemoryUsage().setLimit(1024 * 1024 * 5); activeMQBroker.getSystemUsage().getTempUsage().setLimit(1024 * 1024 * 5); activeMQBroker.start(); + stompClient = new Reactor2TcpStompClient("127.0.0.1", port); stompClient.setMessageConverter(new PassThruMessageConverter()); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.afterPropertiesSet(); stompClient.setTaskScheduler(taskScheduler); + stompClient.setReceiptTimeLimit(5000); } @AfterClass 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 308ec0faff..dc36214cfa 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 @@ -59,6 +59,7 @@ import org.springframework.messaging.simp.broker.SubscriptionRegistry; import org.springframework.messaging.simp.config.MessageBrokerRegistry; import org.springframework.messaging.simp.stomp.StompCommand; import org.springframework.messaging.simp.stomp.StompHeaderAccessor; +import org.springframework.messaging.simp.stomp.StompHeaders; import org.springframework.messaging.support.AbstractSubscribableChannel; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.MessageBuilder; @@ -209,7 +210,6 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests { public WebSocketStompClient stompClient(TaskScheduler taskScheduler) { WebSocketStompClient webSocketStompClient = new WebSocketStompClient(webSocketClient()); webSocketStompClient.setMessageConverter(new MappingJackson2MessageConverter()); - webSocketStompClient.setDefaultHeartbeat(new long[]{100, 100}); webSocketStompClient.setTaskScheduler(taskScheduler); return webSocketStompClient; } @@ -219,6 +219,9 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests { WebSocketStompSessionManager webSocketStompSessionManager = new WebSocketStompSessionManager(stompClient, server().getWsBaseUrl() + "/ws"); webSocketStompSessionManager.setAutoReceipt(true); + StompHeaders stompHeaders = new StompHeaders(); + stompHeaders.setHeartbeat(new long[] {10000, 10000}); + webSocketStompSessionManager.setConnectHeaders(stompHeaders); return webSocketStompSessionManager; } 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 b5b5606837..fd6f59385c 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 @@ -32,6 +32,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; @@ -53,6 +54,7 @@ import org.springframework.integration.stomp.WebSocketStompSessionManager; 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.test.support.LongRunningIntegrationTest; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.websocket.TomcatWebSocketTestServer; import org.springframework.messaging.Message; @@ -95,6 +97,9 @@ import org.springframework.web.socket.sockjs.client.WebSocketTransport; @DirtiesContext public class StompMessageHandlerWebSocketIntegrationTests { + @ClassRule + public static LongRunningIntegrationTest longTests = new LongRunningIntegrationTest(); + @Value("#{server.serverContext}") private ApplicationContext serverContext; @@ -113,10 +118,10 @@ public class StompMessageHandlerWebSocketIntegrationTests { @Test public void testStompMessageHandler() throws InterruptedException { int n = 0; - while (TestUtils.getPropertyValue(this.stompMessageHandler, "stompSession") == null && n++ < 10) { - Thread.sleep(50); + while (TestUtils.getPropertyValue(this.stompMessageHandler, "stompSession") == null && n++ < 100) { + Thread.sleep(100); } - assertTrue(n < 10); + assertTrue(n < 100); StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.SEND); headers.setDestination("/app/simple"); @@ -180,7 +185,7 @@ public class StompMessageHandlerWebSocketIntegrationTests { public WebSocketStompClient stompClient(TaskScheduler taskScheduler) { WebSocketStompClient webSocketStompClient = new WebSocketStompClient(webSocketClient()); webSocketStompClient.setTaskScheduler(taskScheduler); - webSocketStompClient.setReceiptTimeLimit(200); + webSocketStompClient.setReceiptTimeLimit(5000); webSocketStompClient.setMessageConverter(new StringMessageConverter()); return webSocketStompClient; } diff --git a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/test/TestTrigger.java b/spring-integration-test/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java similarity index 81% rename from spring-integration-jpa/src/test/java/org/springframework/integration/jpa/test/TestTrigger.java rename to spring-integration-test/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java index 129f8c736c..dd1156781d 100644 --- a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/test/TestTrigger.java +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-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. You may obtain a copy of the License at @@ -10,7 +10,7 @@ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * specific language governing permissions and limitations under the License. */ -package org.springframework.integration.jpa.test; +package org.springframework.integration.test.util; import java.util.Date; import java.util.concurrent.CountDownLatch; @@ -26,7 +26,7 @@ import org.springframework.scheduling.TriggerContext; * @since 2.2 * */ -public class TestTrigger implements Trigger { +public class OnlyOnceTrigger implements Trigger { private static final AtomicBoolean hasRun = new AtomicBoolean(); @@ -35,15 +35,15 @@ public class TestTrigger implements Trigger { private static volatile CountDownLatch latch = new CountDownLatch(1); - public TestTrigger() { + public OnlyOnceTrigger() { super(); executionTime = new Date(); } public Date nextExecutionTime(TriggerContext triggerContext) { - if (TestTrigger.hasRun.getAndSet(true)) { - TestTrigger.latch.countDown(); + if (OnlyOnceTrigger.hasRun.getAndSet(true)) { + OnlyOnceTrigger.latch.countDown(); return null; } @@ -67,7 +67,7 @@ public class TestTrigger implements Trigger { return false; if (getClass() != obj.getClass()) return false; - TestTrigger other = (TestTrigger) obj; + OnlyOnceTrigger other = (OnlyOnceTrigger) obj; if (executionTime == null) { if (other.executionTime != null) return false; @@ -77,13 +77,13 @@ public class TestTrigger implements Trigger { } public void reset() { - TestTrigger.latch = new CountDownLatch(1); - TestTrigger.hasRun.set(false); + OnlyOnceTrigger.latch = new CountDownLatch(1); + OnlyOnceTrigger.hasRun.set(false); } public void await() { try { - TestTrigger.latch.await(5000, TimeUnit.MILLISECONDS); + OnlyOnceTrigger.latch.await(5000, TimeUnit.MILLISECONDS); if (latch.getCount() != 0) { throw new RuntimeException("test latch.await() did not count down"); }