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

INT-3786: Fix Several Sporadic Test Failures

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3786

* Increase `receiveTimeout` for several `QueueChannel` and `CountDownLatch` based tests
* Fix `FileSplitterTests` for Windows compatibility - `UTF-8` for bytes conversion
* Make some STOMP tests as `LongRunning`

The next fixing phase
This commit is contained in:
Artem Bilan
2015-07-30 11:31:23 -04:00
committed by Gary Russell
parent 3c9b0e4e82
commit ec12b289c2
19 changed files with 111 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java
View File

@@ -308,11 +308,14 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
Iterator<UUID> 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));
}
}
}

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

@@ -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(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<Message<?>> result = template.asyncReceive("testChannel");
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;
assertTrue(elapsed >= 200-safety);
@@ -202,7 +202,7 @@ public class AsyncMessagingTemplateTests {
Future<?> result = template.asyncReceiveAndConvert();
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());
@@ -216,7 +216,7 @@ public class AsyncMessagingTemplateTests {
Future<?> result = template.asyncReceiveAndConvert(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());
@@ -234,7 +234,7 @@ public class AsyncMessagingTemplateTests {
Future<?> result = template.asyncReceiveAndConvert("testChannel");
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;
assertTrue(elapsed >= 200-safety);

20
spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java
View File

@@ -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<File> received = pollableFileSource.receive();
@@ -154,8 +156,10 @@ public class FileReadingMessageSourceIntegrationTests {
}
pollableFileSource.onSend(received);
}
};
Runnable failingConsumer = new Runnable() {
@Override
public void run() {
Message<File> 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;
}
}

10
spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests-context.xml
View File

@@ -1,14 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-file="http://www.springframework.org/schema/integration/file"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd">
<int:splitter input-channel="input1" output-channel="output">
<bean class="org.springframework.integration.file.splitter.FileSplitter">
<constructor-arg value="false"/>
</bean>
</int:splitter>
<int-file:splitter input-channel="input1" output-channel="output" iterator="false" charset="UTF-8"/>
</beans>

8
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CqInboundChannelAdapterTests.java
View File

@@ -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);
}
}
}

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java
View File

@@ -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));
}
}
}

6
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcPollingChannelAdapterParserTests.java
View File

@@ -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() {

21
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java
View File

@@ -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<Object>("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());
}

2
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

@@ -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 {

2
spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java
View File

@@ -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() {

11
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests-context.xml
View File

@@ -1,9 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration" xmlns:context="http://www.springframework.org/schema/context"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:channel id="requests" />
@@ -23,8 +22,12 @@
<property name="locateExistingServerIfPossible" value="true" />
</bean>
<int:inbound-channel-adapter id="source" channel="nullChannel" expression="''">
<int:inbound-channel-adapter id="source" channel="sourceChannel" expression="''">
<int:poller fixed-delay="10" />
</int:inbound-channel-adapter>
<int:channel id="sourceChannel">
<int:queue/>
</int:channel>
</beans>

7
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java
View File

@@ -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);

6
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MonitorTests-context.xml
View File

@@ -16,9 +16,11 @@
<int:channel id="input">
<int:queue />
</int:channel>
<bean id="testTrigger" class="org.springframework.integration.test.util.OnlyOnceTrigger"/>
<int:bridge input-channel="input" output-channel="next">
<int:poller fixed-delay="500" />
<int:poller trigger="testTrigger" />
</int:bridge>
<int:service-activator input-channel="next">

2
spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/BaseJpaPollingChannelAdapterTests-context.xml
View File

@@ -51,7 +51,7 @@
<bean id="consumer" class="org.springframework.integration.jpa.test.Consumer" />
<bean id="testtrigger" class="org.springframework.integration.jpa.test.TestTrigger" />
<bean id="testtrigger" class="org.springframework.integration.test.util.OnlyOnceTrigger" />
<int:poller id="defaultPoller" default="true" fixed-rate="10000" />

28
spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapterTests.java
View File

@@ -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<Collection<?>> 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<Collection<?>> 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<Collection<?>> 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<Collection<?>> 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<Collection<?>> 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<Message<Collection<?>>> received = new ArrayList<Message<Collection<?>>>();
received.add(consumer.poll(5000));
received.add(consumer.poll(10000));
final Message<Collection<?>> message = received.get(0);
@@ -388,7 +390,7 @@ public class JpaPollingChannelAdapterTests {
final Consumer consumer = new Consumer();
final List<Message<Collection<?>>> received = new ArrayList<Message<Collection<?>>>();
received.add(consumer.poll(5000));
received.add(consumer.poll(10000));
final Message<Collection<?>> 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<Collection<?>> 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<Collection<?>> message = received.get(0);

6
spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java
View File

@@ -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

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

@@ -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;
}

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

@@ -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;
}

20
spring-integration-jpa/src/test/java/org/springframework/integration/jpa/test/TestTrigger.java → spring-integration-test/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java
View File

@@ -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");
}