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

INT-3729: Fix STOMP test sporadic failures

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3729
This commit is contained in:
Artem Bilan
2015-06-07 20:17:01 -04:00
parent 9e1119b434
commit b7221938af
5 changed files with 58 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-core/src/main/java/org/springframework/integration/support/converter/PassThruMessageConverter.java
View File

@@ -23,12 +23,13 @@ import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
/**
* The simple {@link MessageConverter} implementation which contact is to return
* The simple {@link MessageConverter} implementation which contract is to return
* {@link Message} as is for both {@code from/to} operations.
* <p>
* It is useful in cases of some protocol implementations (e.g. STOMP),
* which is based on the "Spring Messaging Foundation" and the further logic
* operates only with {@link Message}s, e.g. Spring Integration Adapters.
*
* @author Artem Bilan
* @since 4.2
*/

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

@@ -146,25 +146,9 @@ import reactor.spring.context.config.EnableReactor;
@ContextConfiguration(loader = AnnotationConfigContextLoader.class,
classes = {EnableIntegrationTests.ContextConfiguration.class, EnableIntegrationTests.ContextConfiguration2.class})
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
public class EnableIntegrationTests {
private final Log logger = LogFactory.getLog(EnableIntegrationTests.class);
@Rule
public TestRule watcher = new TestWatcher() {
protected void starting(Description description) {
logger.debug("STARTING TEST: " + description.getMethodName());
}
@Override
protected void finished(Description description) {
logger.debug("FINISHED TEST: " + description.getMethodName());
}
};
@Autowired
private ApplicationContext context;
@@ -283,20 +267,6 @@ public class EnableIntegrationTests {
@Autowired
private MessageChannel controlBusChannel;
private static Level existingLogLevel;
// The temporal hooks to investigate CI failures
@BeforeClass
public static void setup() {
existingLogLevel = LogManager.getLogger("org.springframework.integration").getLevel();
LogManager.getLogger("org.springframework.integration").setLevel(Level.DEBUG);
}
@AfterClass
public static void tearDown() {
LogManager.getLogger("org.springframework.integration").setLevel(existingLogLevel);
}
@Test
public void testAnnotatedServiceActivator() {
assertEquals(10L, TestUtils.getPropertyValue(this.serviceActivatorEndpoint, "maxMessagesPerPoll"));
@@ -337,9 +307,6 @@ public class EnableIntegrationTests {
assertEquals(10L, TestUtils.getPropertyValue(trigger, "period"));
assertFalse(TestUtils.getPropertyValue(trigger, "fixedRate", Boolean.class));
// Markers to investigate the failures on CI
logger.debug("----SEND Message to 'input' channel----");
this.input.send(MessageBuilder.withPayload("Foo").build());
Message<?> interceptedMessage = this.wireTapChannel.receive(10000);
@@ -360,8 +327,6 @@ public class EnableIntegrationTests {
receive = this.publishedChannel.receive(10000);
logger.debug("----RECEIVE Message from 'publishedChannel' channel----" + receive);
assertNotNull(receive);
assertEquals("foo", receive.getPayload());
@@ -417,7 +382,6 @@ public class EnableIntegrationTests {
}
@Test
@DirtiesContext
public void testChangePatterns() {
try {
this.configurer.setComponentNamePatterns(new String[] {"*"});
@@ -729,10 +693,7 @@ public class EnableIntegrationTests {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
fbInterceptorCounter().incrementAndGet();
Message<?> message1 = super.preSend(message, channel);
logger.debug("!!!!'ciFactoryBean': the result of 'preSend' on '" + channel + "' '" +
message1 + "'");
return message1;
return super.preSend(message, channel);
}
};
}
@@ -846,8 +807,6 @@ public class EnableIntegrationTests {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
this.invoked.incrementAndGet();
logger.debug("!!!!'TestChannelInterceptor': the result of 'preSend' on '" + channel + "' '" +
message + "'");
return message;
}
@@ -947,52 +906,9 @@ public class EnableIntegrationTests {
};
}
@Bean(name = IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)
public TaskScheduler taskScheduler(BeanFactory beanFactory) {
@SuppressWarnings("serial")
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler() {
@Override
public void afterPropertiesSet() {
logger.debug("INITIALIZING taskScheduler...");
super.afterPropertiesSet();
}
@Override
public void destroy() {
logger.debug("DESTROYING taskScheduler...");
super.destroy();
}
};
taskScheduler.setPoolSize(20);
taskScheduler.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
taskScheduler.setThreadNamePrefix("my-task-scheduler-");
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
errorHandler.setBeanFactory(beanFactory);
taskScheduler.setErrorHandler(errorHandler);
return taskScheduler;
}
@Bean
public PollableChannel publishedChannel() {
return new QueueChannel() {
@Override
protected boolean doSend(Message<?> message, long timeout) {
logger.debug("---- 'publishedChannel' Thread State: " + Thread.currentThread().isInterrupted());
logger.debug("---- 'publishedChannel' before 'doSend': " + message);
logger.debug("---- 'publishedChannel' state before: " + getQueueSize() +
", " + getRemainingCapacity());
boolean b = super.doSend(message, timeout);
logger.debug("---- 'publishedChannel' after 'doSend': " + b);
logger.debug("---- 'publishedChannel' state after: " + getQueueSize() +
", " + getRemainingCapacity());
return b;
}
};
return new QueueChannel();
}
@Bean

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

@@ -41,7 +41,7 @@ import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.util.Assert;
/**
* The {@link AbstractMessageHandler} implemntation to send messages to STOMP destinations.
* The {@link AbstractMessageHandler} implementation to send messages to STOMP destinations.
*
* @author Artem Bilan
* @since 4.2

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

@@ -16,15 +16,22 @@
package org.springframework.integration.stomp.client;
import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
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;
import org.junit.Test;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -33,8 +40,11 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
import org.springframework.integration.stomp.Reactor2TcpStompSessionManager;
import org.springframework.integration.stomp.StompSessionManager;
import org.springframework.integration.stomp.event.StompIntegrationEvent;
import org.springframework.integration.stomp.event.StompReceiptEvent;
import org.springframework.integration.stomp.inbound.StompInboundChannelAdapter;
import org.springframework.integration.stomp.outbound.StompMessageHandler;
import org.springframework.integration.support.converter.PassThruMessageConverter;
@@ -43,7 +53,11 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
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;
/**
@@ -69,6 +83,9 @@ public class StompServerIntegrationTests {
activeMQBroker.start();
stompClient = new Reactor2TcpStompClient("127.0.0.1", port);
stompClient.setMessageConverter(new PassThruMessageConverter());
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.afterPropertiesSet();
stompClient.setTaskScheduler(taskScheduler);
}
@AfterClass
@@ -81,12 +98,29 @@ public class StompServerIntegrationTests {
ConfigurableApplicationContext context1 = new AnnotationConfigApplicationContext(ContextConfiguration.class);
ConfigurableApplicationContext context2 = new AnnotationConfigApplicationContext(ContextConfiguration.class);
PollableChannel stompEvents1 = context1.getBean("stompEvents", PollableChannel.class);
PollableChannel stompEvents2 = context2.getBean("stompEvents", PollableChannel.class);
PollableChannel stompInputChannel1 = context1.getBean("stompInputChannel", PollableChannel.class);
PollableChannel stompInputChannel2 = context2.getBean("stompInputChannel", PollableChannel.class);
MessageChannel stompOutputChannel1 = context1.getBean("stompOutputChannel", MessageChannel.class);
MessageChannel stompOutputChannel2 = context2.getBean("stompOutputChannel", MessageChannel.class);
Message<?> eventMessage = stompEvents1.receive(10000);
assertNotNull(eventMessage);
assertThat(eventMessage.getPayload(), instanceOf(StompReceiptEvent.class));
StompReceiptEvent stompReceiptEvent = (StompReceiptEvent) eventMessage.getPayload();
assertEquals(StompCommand.SUBSCRIBE, stompReceiptEvent.getStompCommand());
assertEquals("/topic/myTopic", stompReceiptEvent.getDestination());
eventMessage = stompEvents2.receive(10000);
assertNotNull(eventMessage);
assertThat(eventMessage.getPayload(), instanceOf(StompReceiptEvent.class));
stompReceiptEvent = (StompReceiptEvent) eventMessage.getPayload();
assertEquals(StompCommand.SUBSCRIBE, stompReceiptEvent.getStompCommand());
assertEquals("/topic/myTopic", stompReceiptEvent.getDestination());
stompOutputChannel1.send(new GenericMessage<byte[]>("Hello, Client#2!".getBytes()));
Message<?> receive11 = stompInputChannel1.receive(10000);
@@ -138,7 +172,9 @@ public class StompServerIntegrationTests {
@Bean
public StompSessionManager stompSessionManager() {
return new Reactor2TcpStompSessionManager(stompClient);
Reactor2TcpStompSessionManager stompSessionManager = new Reactor2TcpStompSessionManager(stompClient);
stompSessionManager.setAutoReceipt(true);
return stompSessionManager;
}
@Bean
@@ -162,6 +198,20 @@ public class StompServerIntegrationTests {
return handler;
}
@Bean
public PollableChannel stompEvents() {
return new QueueChannel();
}
@Bean
@SuppressWarnings("unchecked")
public ApplicationListener<ApplicationEvent> stompEventListener() {
ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer();
producer.setEventTypes(StompIntegrationEvent.class);
producer.setOutputChannel(stompEvents());
return producer;
}
}
}

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

@@ -209,6 +209,7 @@ 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;
}