From 353e99aa80473e2a9f2282b252662c29d2c25256 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 22 Jun 2021 17:05:46 -0400 Subject: [PATCH] Rework ZeroMqMessageProducerTests for XPUB * Use an `XPUB` socket to receive subscriptions before publishing. This makes the test more robust and less blocked for that `Thread.sleep()` --- .../zeromq/inbound/ZeroMqMessageProducerTests.java | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java index 58f7385eee..586b8c7547 100644 --- a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java @@ -94,10 +94,11 @@ public class ZeroMqMessageProducerTests { } @Test - void testMessageProducerForPubSubReceiveRaw() throws InterruptedException { + void testMessageProducerForPubSubReceiveRaw() { String socketAddress = "inproc://messageProducer.test"; - ZMQ.Socket socket = CONTEXT.createSocket(SocketType.PUB); + ZMQ.Socket socket = CONTEXT.createSocket(SocketType.XPUB); socket.bind(socketAddress); + socket.setReceiveTimeOut(10_000); FluxMessageChannel outputChannel = new FluxMessageChannel(); @@ -126,8 +127,7 @@ public class ZeroMqMessageProducerTests { messageProducer.afterPropertiesSet(); messageProducer.start(); - // Give it some time to connect and subscribe - Thread.sleep(2000); + assertThat(socket.recv()).isNotNull(); ZMsg msg = ZMsg.newStringMsg("test"); msg.wrap(new ZFrame("testTopic")); @@ -135,8 +135,7 @@ public class ZeroMqMessageProducerTests { messageProducer.subscribeToTopics("other"); - // Give it some time to connect and subscribe - Thread.sleep(2000); + assertThat(socket.recv()).isNotNull(); msg = ZMsg.newStringMsg("test"); msg.wrap(new ZFrame("otherTopic"));