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

Fix one more STOMP WebSocket test race condition

Browse Source 
https://build.spring.io/browse/INT-B43-JOB1-7

We need to wait for the **real** `UNSUBSCRIBE` from server before go ahead.
This commit is contained in:
Artem Bilan
2015-11-19 19:23:10 -05:00
parent e55a7b8234
commit 350355e362
1 changed files with 19 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -151,8 +151,10 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju
this.stompInboundChannelAdapter.removeDestination("/topic/myTopic"); this.stompInboundChannelAdapter.removeDestination("/topic/myTopic");
waitForUnsubscribe("/topic/myTopic");
messagingTemplate.convertAndSend("/topic/myTopic", "foo"); messagingTemplate.convertAndSend("/topic/myTopic", "foo");
receive = this.stompInputChannel.receive(1000); receive = this.errorChannel.receive(100);
assertNull(receive); assertNull(receive);
this.stompInboundChannelAdapter.addDestination("/topic/myTopic"); this.stompInboundChannelAdapter.addDestination("/topic/myTopic");
@@ -191,6 +193,7 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju
messagingTemplate.convertAndSend("/topic/myTopic", "foo"); messagingTemplate.convertAndSend("/topic/myTopic", "foo");
receive = this.errorChannel.receive(10000); receive = this.errorChannel.receive(10000);
assertNotNull(receive); assertNotNull(receive);
assertEquals(0, ((QueueChannel) this.errorChannel).getQueueSize());
} }
private void waitForSubscribe(String destination) throws InterruptedException { private void waitForSubscribe(String destination) throws InterruptedException {
@@ -207,6 +210,21 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju
assertTrue("The subscription for the '" + destination + "' destination hasn't been registered", n < 100); assertTrue("The subscription for the '" + destination + "' destination hasn't been registered", n < 100);
} }
private void waitForUnsubscribe(String destination) throws InterruptedException {
SimpleBrokerMessageHandler serverBrokerMessageHandler =
this.serverContext.getBean("simpleBrokerMessageHandler", SimpleBrokerMessageHandler.class);
SubscriptionRegistry subscriptionRegistry = serverBrokerMessageHandler.getSubscriptionRegistry();
int n = 0;
while (containsDestination(destination, subscriptionRegistry) && n++ < 100) {
Thread.sleep(100);
}
assertTrue("The subscription for the '" + destination + "' destination hasn't been registered", n < 100);
}
private boolean containsDestination(String destination, SubscriptionRegistry subscriptionRegistry) { private boolean containsDestination(String destination, SubscriptionRegistry subscriptionRegistry) {
StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.create(StompCommand.MESSAGE); StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.create(StompCommand.MESSAGE);
stompHeaderAccessor.setDestination(destination); stompHeaderAccessor.setDestination(destination);