From 60b355bc3852511751ae926ebe69f1dca6b4f77b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 23 Jun 2021 10:33:28 -0400 Subject: [PATCH] Move subscription to PUB after connection Looks like `subSocket.subscribe("test");` is lost when there is no other side of PUB/SUB connection --- .../outbound/ZeroMqMessageHandlerTests.java | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 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 edfc63034d..ecb1a8b546 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 @@ -90,7 +90,6 @@ public class ZeroMqMessageHandlerTests { ZMQ.Socket subSocket = CONTEXT.createSocket(SocketType.SUB); subSocket.setReceiveTimeOut(20_000); int port = subSocket.bindToRandomPort("tcp://*"); - subSocket.subscribe("test"); ZeroMqMessageHandler messageHandler = new ZeroMqMessageHandler(CONTEXT, "tcp://localhost:" + port, SocketType.PUB); @@ -104,14 +103,16 @@ public class ZeroMqMessageHandlerTests { await().atMost(Duration.ofSeconds(20)).pollDelay(Duration.ofMillis(100)) .untilAsserted(() -> { - messageHandler.handleMessage(testMessage).subscribe(); - ZMsg msg = ZMsg.recvMsg(subSocket); - assertThat(msg).isNotNull(); - assertThat(msg.unwrap().getString(ZMQ.CHARSET)).isEqualTo("testTopic"); - Message capturedMessage = new EmbeddedJsonHeadersMessageMapper().toMessage(msg.getFirst().getData()); - assertThat(capturedMessage).isEqualTo(testMessage); - msg.destroy(); - }); + subSocket.subscribe("test"); + messageHandler.handleMessage(testMessage).subscribe(); + ZMsg msg = ZMsg.recvMsg(subSocket); + assertThat(msg).isNotNull(); + assertThat(msg.unwrap().getString(ZMQ.CHARSET)).isEqualTo("testTopic"); + Message capturedMessage = + new EmbeddedJsonHeadersMessageMapper().toMessage(msg.getFirst().getData()); + assertThat(capturedMessage).isEqualTo(testMessage); + msg.destroy(); + }); messageHandler.destroy(); subSocket.close();