Remove SocketUtils usage
* Remove usage of non-stable `org.springframework.util.SocketUtils` * Replace it with `0` for those tests where it is possible to select OS port * Remove the mentioning of the `SocketUtils` from the `testing.adoc` * Use `TransportConstants.DEFAULT_STOMP_PORT` for `StompServerIntegrationTests`. We may disable this test in the future for CI if it is not going to be stable * Introduce `Supplier<String> connectUrl` variants for `ZeroMqMessageHandler` to let it defer connection evaluation until subscription to the socket `Mono` in the `ZeroMqMessageHandler`. * Move connection logic in the `ZeroMqMessageHandler` to `Lifecycle.start()` Related to https://github.com/spring-projects/spring-framework/issues/28054 **Cherry-pick to `5.5.x`** # Conflicts: # spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2020 the original author or authors.
|
||||
* Copyright 2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -44,7 +44,6 @@ import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
|
||||
import org.springframework.util.SocketUtils;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
@@ -55,8 +54,6 @@ import org.springframework.util.SocketUtils;
|
||||
@DirtiesContext
|
||||
public class ZeroMqDslTests {
|
||||
|
||||
private static final int PROXY_PUB_PORT = SocketUtils.findAvailableTcpPort();
|
||||
|
||||
@Autowired
|
||||
ZContext context;
|
||||
|
||||
@@ -82,10 +79,10 @@ public class ZeroMqDslTests {
|
||||
for (int i = 0; i < 2; i++) {
|
||||
IntegrationFlow consumerFlow =
|
||||
IntegrationFlows.from(
|
||||
ZeroMq.inboundChannelAdapter(this.context, SocketType.SUB)
|
||||
.connectUrl("tcp://localhost:" + this.subPubZeroMqProxy.getBackendPort())
|
||||
.topics("someTopic")
|
||||
.consumeDelay(Duration.ofMillis(100)))
|
||||
ZeroMq.inboundChannelAdapter(this.context, SocketType.SUB)
|
||||
.connectUrl("tcp://localhost:" + this.subPubZeroMqProxy.getBackendPort())
|
||||
.topics("someTopic")
|
||||
.consumeDelay(Duration.ofMillis(100)))
|
||||
.channel(ZeroMq.zeroMqChannel(this.context).zeroMqProxy(this.pullPushZeroMqProxy))
|
||||
.transform(Transformers.objectToString())
|
||||
.handle(results::offer)
|
||||
@@ -130,9 +127,7 @@ public class ZeroMqDslTests {
|
||||
|
||||
@Bean
|
||||
ZeroMqProxy subPubZeroMqProxy() {
|
||||
ZeroMqProxy zeroMqProxy = new ZeroMqProxy(context(), ZeroMqProxy.Type.SUB_PUB);
|
||||
zeroMqProxy.setFrontendPort(PROXY_PUB_PORT);
|
||||
return zeroMqProxy;
|
||||
return new ZeroMqProxy(context(), ZeroMqProxy.Type.SUB_PUB);
|
||||
}
|
||||
|
||||
@Bean
|
||||
@@ -141,10 +136,14 @@ public class ZeroMqDslTests {
|
||||
}
|
||||
|
||||
@Bean
|
||||
IntegrationFlow publishToZeroMqPubSubFlow() {
|
||||
IntegrationFlow publishToZeroMqPubSubFlow(ZeroMqProxy subPubZeroMqProxy) {
|
||||
return flow ->
|
||||
flow.handle(ZeroMq.outboundChannelAdapter(context(), "tcp://localhost:" + PROXY_PUB_PORT,
|
||||
SocketType.PUB)
|
||||
flow.handle(ZeroMq.outboundChannelAdapter(context(),
|
||||
() -> {
|
||||
await().until(() -> subPubZeroMqProxy.getFrontendPort() > 0);
|
||||
return "tcp://localhost:" + subPubZeroMqProxy.getFrontendPort();
|
||||
},
|
||||
SocketType.PUB)
|
||||
.topic("someTopic"));
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2020-2021 the original author or authors.
|
||||
* Copyright 2020-2022 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -71,6 +71,7 @@ public class ZeroMqMessageHandlerTests {
|
||||
Mono<ZMQ.Socket> socketMono = TestUtils.getPropertyValue(messageHandler, "socketMono", Mono.class);
|
||||
ZMQ.Socket socketInUse = socketMono.block(Duration.ofSeconds(10));
|
||||
assertThat(socketInUse.getZapDomain()).isEqualTo("global");
|
||||
messageHandler.start();
|
||||
|
||||
Message<?> testMessage = new GenericMessage<>("test");
|
||||
messageHandler.handleMessage(testMessage).subscribe();
|
||||
@@ -99,6 +100,7 @@ public class ZeroMqMessageHandlerTests {
|
||||
new FunctionExpression<Message<?>>((message) -> message.getHeaders().get("topic")));
|
||||
messageHandler.setMessageMapper(new EmbeddedJsonHeadersMessageMapper());
|
||||
messageHandler.afterPropertiesSet();
|
||||
messageHandler.start();
|
||||
|
||||
Message<?> testMessage = MessageBuilder.withPayload("test").setHeader("topic", "testTopic").build();
|
||||
|
||||
@@ -137,6 +139,7 @@ public class ZeroMqMessageHandlerTests {
|
||||
messageHandler.setBeanFactory(mock(BeanFactory.class));
|
||||
messageHandler.setMessageConverter(new ByteArrayMessageConverter());
|
||||
messageHandler.afterPropertiesSet();
|
||||
messageHandler.start();
|
||||
|
||||
Message<?> testMessage = new GenericMessage<>("test".getBytes());
|
||||
messageHandler.handleMessage(testMessage).subscribe();
|
||||
|
||||
Reference in New Issue
Block a user