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.

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