From 046875de7ffc049f987040a73205f7beb120491f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Sun, 8 Feb 2015 16:31:53 +0200 Subject: [PATCH] INT-3580: fix race condition for `StompIntTests` JIRA: https://jira.spring.io/browse/INT-3580 The test `StompIntegrationTests#handleExceptionAndSendToUser` sends `SUBSCRIBE` and `MESSAGE` messages at once. Since Spring Websocket support relies on the `ExecutorSubscribableChannel` for `clientInboundChannel`, there is no guaranty that messages are processed with the same order as they have been sent. From other side `UserDestinationMessageHandler` shifts messages to the `brokerChannel`, which is an another `ExecutorSubscribableChannel` with its own `Executor`. Hence there is some race condition when the second message can be handled before the first one. Add `check subscription` cycle to wait until the `SUBSCRIBE` message registers its subscription to the Broker. And only after that send the `MESSAGE` message. --- .../client/StompIntegrationTests.java | 27 ++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java index a26fa5440a..95cdd202ed 100644 --- a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-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. @@ -25,6 +25,7 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; import java.nio.ByteBuffer; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -49,6 +50,7 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.transformer.ExpressionEvaluatingTransformer; import org.springframework.integration.websocket.ClientWebSocketContainer; import org.springframework.integration.websocket.IntegrationWebSocketContainer; @@ -63,6 +65,8 @@ import org.springframework.messaging.handler.annotation.MessageExceptionHandler; import org.springframework.messaging.handler.annotation.MessageMapping; import org.springframework.messaging.simp.annotation.SendToUser; import org.springframework.messaging.simp.annotation.SubscribeMapping; +import org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler; +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; @@ -71,7 +75,6 @@ import org.springframework.stereotype.Controller; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import org.springframework.web.socket.client.jetty.JettyWebSocketClient; import org.springframework.web.socket.client.standard.StandardWebSocketClient; import org.springframework.web.socket.config.annotation.AbstractWebSocketMessageBrokerConfigurer; import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; @@ -205,6 +208,24 @@ public class StompIntegrationTests { Message message2 = MessageBuilder.withPayload("foo").setHeaders(headers).build(); this.webSocketOutputChannel.send(message); + + SimpleBrokerMessageHandler serverBrokerMessageHandler = + this.serverContext.getBean("simpleBrokerMessageHandler", SimpleBrokerMessageHandler.class); + + SubscriptionRegistry subscriptionRegistry = serverBrokerMessageHandler.getSubscriptionRegistry(); + + @SuppressWarnings("rawtypes") + Map subscriptions = TestUtils.getPropertyValue(subscriptionRegistry, "subscriptionRegistry.sessions", Map.class); + + int n = 0; + + while (subscriptions.isEmpty() && n++ < 100) { + Thread.sleep(100); + subscriptions = TestUtils.getPropertyValue(subscriptionRegistry, "subscriptionRegistry.sessions", Map.class); + } + + assertTrue("The subscription for the 'user/queue/error' hasn't been registered", n < 100); + this.webSocketOutputChannel.send(message2); @@ -301,7 +322,7 @@ public class StompIntegrationTests { @IntegrationTestController static class SimpleController { - private CountDownLatch latch = new CountDownLatch(1); + private final CountDownLatch latch = new CountDownLatch(1); @MessageMapping(value = "/simple") public void handle() {