From d7455e7f27cf7288abe3b85e783f18d1aedb2276 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 18 Nov 2015 13:53:50 -0500 Subject: [PATCH] INT-3888: Fix STOMP Test for Race Condition 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 --- .../stomp/AbstractStompSessionManager.java | 7 ++--- .../client/StompServerIntegrationTests.java | 2 +- ...annelAdapterWebSocketIntegrationTests.java | 27 +++++++------------ 3 files changed, 12 insertions(+), 24 deletions(-) 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 0b3e85f213..95158b3b0d 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 @@ -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); 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 13450ddcb1..00950da56e 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 @@ -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)); 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 da35e7b1de..5497920cc3 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 @@ -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 message = MessageBuilder.createMessage(new byte[0], stompHeaderAccessor.toMessageHeaders()); + MultiValueMap subscriptions = subscriptionRegistry.findSubscriptions(message); + return !subscriptions.isEmpty(); }