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

INT-3888: Fix STOMP Test for Race Condition

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

Note sure yet what race condition we have in the `DefaultSubscriptionRegistry`,
but that looks for me better to follow with the standard `SubscriptionRegistry.findSubscriptions()`
to check subscription present, unless fail during `send()` because of `destinationLookup` early exit.

Fix for the `StompServerIntegrationTests`

Looks like we have to `reconnect()` for any `handleTransportError()` activity,
otherwise we are just hang in the Reactor's TCP reconnect loop without freeing resources
for other activities, e.g. start Embedded STOMP Broker.

Increase `receive` timeout in case of wait for the Broker Start
This commit is contained in:
Artem Bilan
2015-11-18 13:53:50 -05:00
committed by Gary Russell
parent 1fe50138f0
commit d7455e7f27
3 changed files with 12 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -32,7 +32,6 @@ import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.stomp.event.StompConnectionFailedEvent;
import org.springframework.integration.stomp.event.StompSessionConnectedEvent;
import org.springframework.messaging.simp.stomp.ConnectionLostException;
import org.springframework.messaging.simp.stomp.StompClientSupport;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaders;
@@ -352,10 +351,8 @@ 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);
}
this.session = null;
scheduleReconnect(exception);
synchronized (this.delegates) {
for (StompSessionHandler delegate : this.delegates) {
delegate.handleTransportError(session, exception);

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

@@ -214,7 +214,7 @@ public class StompServerIntegrationTests extends LogAdjustingTestSupport {
activeMQBroker.start(false);
do {
eventMessage = stompEvents1.receive(10000);
eventMessage = stompEvents1.receive(20000);
assertNotNull(eventMessage);
}
while (!(eventMessage.getPayload() instanceof StompReceiptEvent));

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

@@ -27,8 +27,6 @@ import static org.junit.Assert.assertTrue;
import java.util.Collections;
import java.util.Map;
import org.apache.log4j.Level;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -50,9 +48,7 @@ import org.springframework.integration.stomp.event.StompConnectionFailedEvent;
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;
import org.springframework.messaging.MessageHandlingException;
@@ -73,6 +69,7 @@ import org.springframework.scheduling.TaskScheduler;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.MultiValueMap;
import org.springframework.web.socket.client.WebSocketClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.config.annotation.AbstractWebSocketMessageBrokerConfigurer;
@@ -130,7 +127,7 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju
assertEquals(StompCommand.SUBSCRIBE, stompReceiptEvent.getStompCommand());
assertEquals("/topic/myTopic", stompReceiptEvent.getDestination());
waitForSubscribe("myTopic");
waitForSubscribe("/topic/myTopic");
SimpMessagingTemplate messagingTemplate = this.serverContext.getBean("brokerMessagingTemplate",
SimpMessagingTemplate.class);
@@ -162,7 +159,7 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju
receive = this.stompEvents.receive(10000);
assertNotNull(receive);
waitForSubscribe("myTopic");
waitForSubscribe("/topic/myTopic");
messagingTemplate.convertAndSend("/topic/myTopic", "foo");
receive = this.errorChannel.receive(10000);
@@ -188,7 +185,7 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju
}
while (!(eventMessage.getPayload() instanceof StompSessionConnectedEvent));
waitForSubscribe("myTopic");
waitForSubscribe("/topic/myTopic");
messagingTemplate = this.serverContext.getBean("brokerMessagingTemplate", SimpMessagingTemplate.class);
messagingTemplate.convertAndSend("/topic/myTopic", "foo");
@@ -210,18 +207,12 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju
assertTrue("The subscription for the '" + destination + "' destination hasn't been registered", n < 100);
}
@SuppressWarnings("rawtypes")
private boolean containsDestination(String destination, SubscriptionRegistry subscriptionRegistry) {
Map sessions = TestUtils.getPropertyValue(subscriptionRegistry, "subscriptionRegistry.sessions", Map.class);
for (Object info : sessions.values()) {
Map subscriptions = TestUtils.getPropertyValue(info, "destinationLookup", Map.class);
for (Object dest : subscriptions.keySet()) {
if (((String) dest).contains(destination)) {
return true;
}
}
}
return false;
StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.create(StompCommand.MESSAGE);
stompHeaderAccessor.setDestination(destination);
Message<byte[]> message = MessageBuilder.createMessage(new byte[0], stompHeaderAccessor.toMessageHeaders());
MultiValueMap<String, String> subscriptions = subscriptionRegistry.findSubscriptions(message);
return !subscriptions.isEmpty();
}