From 0af3d503d906d595adde8a69f51c5221b89919b4 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 19 Nov 2015 19:23:10 -0500 Subject: [PATCH] Fix one more STOMP WebSocket test race condition https://build.spring.io/browse/INT-B43-JOB1-7 We need to wait for the **real** `UNSUBSCRIBE` from server before go ahead. (cherry picked from commit 350355e) --- ...annelAdapterWebSocketIntegrationTests.java | 20 ++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) 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 5497920cc3..88224083d7 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 @@ -151,8 +151,10 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju this.stompInboundChannelAdapter.removeDestination("/topic/myTopic"); + waitForUnsubscribe("/topic/myTopic"); + messagingTemplate.convertAndSend("/topic/myTopic", "foo"); - receive = this.stompInputChannel.receive(1000); + receive = this.errorChannel.receive(100); assertNull(receive); this.stompInboundChannelAdapter.addDestination("/topic/myTopic"); @@ -191,6 +193,7 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju messagingTemplate.convertAndSend("/topic/myTopic", "foo"); receive = this.errorChannel.receive(10000); assertNotNull(receive); + assertEquals(0, ((QueueChannel) this.errorChannel).getQueueSize()); } 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); } + 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) { StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.create(StompCommand.MESSAGE); stompHeaderAccessor.setDestination(destination);