From 39c3acb042c26f0141dd7c1e0d344acb2dbc5f62 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 9 Jul 2021 11:38:03 -0400 Subject: [PATCH] Fix ZMQMessageHandlerTests race condition Use `receiveTimeOut = 0` on SUB socket and rely on the retries from the `await().atMost()`. Also resubscribe in the retry callback before publishing the next attempt --- .../zeromq/outbound/ZeroMqMessageHandlerTests.java | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java index 8935e124b5..47f18a1d93 100644 --- a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java @@ -86,10 +86,11 @@ public class ZeroMqMessageHandlerTests { } @Test - void testMessageHandlerForPubSub() throws InterruptedException { + void testMessageHandlerForPubSub() { ZMQ.Socket subSocket = CONTEXT.createSocket(SocketType.SUB); - subSocket.setReceiveTimeOut(20_000); + subSocket.setReceiveTimeOut(0); int port = subSocket.bindToRandomPort("tcp://*"); + subSocket.subscribe("test"); ZeroMqMessageHandler messageHandler = new ZeroMqMessageHandler(CONTEXT, "tcp://localhost:" + port, SocketType.PUB); @@ -98,15 +99,12 @@ public class ZeroMqMessageHandlerTests { new FunctionExpression>((message) -> message.getHeaders().get("topic"))); messageHandler.setMessageMapper(new EmbeddedJsonHeadersMessageMapper()); messageHandler.afterPropertiesSet(); - subSocket.subscribe("test"); - - // Give it some time to connect and subscribe - Thread.sleep(2000); Message testMessage = MessageBuilder.withPayload("test").setHeader("topic", "testTopic").build(); await().atMost(Duration.ofSeconds(20)).pollDelay(Duration.ofMillis(100)) .untilAsserted(() -> { + subSocket.subscribe("test"); messageHandler.handleMessage(testMessage).subscribe(); ZMsg msg = ZMsg.recvMsg(subSocket); assertThat(msg).isNotNull();