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

Move subscription to PUB after connection

Browse Source 
Looks like `subSocket.subscribe("test");` is lost when there is no
other side of PUB/SUB connection
This commit is contained in:
Artem Bilan
2021-06-23 10:33:28 -04:00
parent 353e99aa80
commit 60b355bc38
1 changed files with 10 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java
View File

@@ -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();