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

INT-3580: Fix More Stomp Test Race Conditions

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3580
This commit is contained in:
Gary Russell
2015-02-12 17:49:00 -05:00
parent b7d64d12da
commit 4286fc6b82
1 changed files with 29 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java
View File

@@ -134,6 +134,9 @@ public class StompIntegrationTests {
Message<Integer> message2 = MessageBuilder.withPayload(5).setHeaders(headers).build();
this.webSocketOutputChannel.send(message);
waitForSubscribe();
this.webSocketOutputChannel.send(message2);
Message<?> receive = webSocketInputChannel.receive(10000);
@@ -157,6 +160,9 @@ public class StompIntegrationTests {
Message<Integer> message2 = MessageBuilder.withPayload(10).setHeaders(headers).build();
this.webSocketOutputChannel.send(message);
waitForSubscribe();
this.webSocketOutputChannel.send(message2);
Message<?> receive = webSocketInputChannel.receive(10000);
@@ -209,22 +215,7 @@ public class StompIntegrationTests {
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);
waitForSubscribe();
this.webSocketOutputChannel.send(message2);
@@ -256,6 +247,9 @@ public class StompIntegrationTests {
Message<String> message2 = MessageBuilder.withPayload("Bob").setHeaders(headers).build();
this.webSocketOutputChannel.send(message);
waitForSubscribe();
this.webSocketOutputChannel.send(message2);
Message<?> receive = webSocketInputChannel.receive(10000);
@@ -263,6 +257,25 @@ public class StompIntegrationTests {
assertEquals("Hello Bob", receive.getPayload());
}
private void waitForSubscribe() throws InterruptedException {
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);
}
@Configuration
@EnableIntegration