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

Optimize ZeroMqMessageHandlerTests for ZMQ.Poller

Browse Source 
* Fix unused import in the `IntegrationDynamicWebSocketHandlerMapping`
This commit is contained in:
Artem Bilan
2021-06-17 11:21:57 -04:00
parent 93743f69fe
commit d7dfea7bd0
2 changed files with 18 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/IntegrationDynamicWebSocketHandlerMapping.java
View File

@@ -28,14 +28,13 @@ import org.springframework.http.server.PathContainer;
import org.springframework.http.server.RequestPath;
import org.springframework.web.HttpRequestHandler;
import org.springframework.web.servlet.HandlerExecutionChain;
import org.springframework.web.servlet.handler.AbstractHandlerMapping;
import org.springframework.web.servlet.handler.AbstractUrlHandlerMapping;
import org.springframework.web.util.ServletRequestPathUtils;
import org.springframework.web.util.pattern.PathPattern;
import org.springframework.web.util.pattern.PathPatternParser;
/**
* The {@link AbstractHandlerMapping} implementation for dynamic WebSocket endpoint registrations in Spring Integration.
* The {@link AbstractUrlHandlerMapping} implementation for dynamic WebSocket endpoint registrations in Spring Integration.
* <p>
* TODO until https://github.com/spring-projects/spring-framework/issues/26798
*

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

@@ -86,7 +86,7 @@ public class ZeroMqMessageHandlerTests {
}
@Test
void testMessageHandlerForPubSub() throws InterruptedException {
void testMessageHandlerForPubSub() {
ZMQ.Socket subSocket = CONTEXT.createSocket(SocketType.SUB);
subSocket.setReceiveTimeOut(20_000);
int port = subSocket.bindToRandomPort("tcp://*");
@@ -100,19 +100,27 @@ public class ZeroMqMessageHandlerTests {
messageHandler.setMessageMapper(new EmbeddedJsonHeadersMessageMapper());
messageHandler.afterPropertiesSet();
// Give it some time to bind and subscribe
Thread.sleep(2000);
ZMQ.Poller poller = CONTEXT.createPoller(1);
poller.register(subSocket, ZMQ.Poller.POLLIN);
Message<?> testMessage = MessageBuilder.withPayload("test").setHeader("topic", "testTopic").build();
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);
while (true) {
poller.poll(10000);
if (poller.pollin(0)) {
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();
break;
}
}
msg.destroy();
poller.unregister(subSocket);
poller.close();
messageHandler.destroy();
subSocket.close();
}