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

Stop using SocketUtils in ActiveMQ STOMP broker tests

Browse Source 
See gh-28052
This commit is contained in:
Sam Brannen
2022-02-15 15:53:43 +01:00
parent 5fbd01f28f
commit bbd35ded91
2 changed files with 27 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/ReactorNettyTcpStompClientTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.messaging.simp.stomp;
import java.lang.reflect.Type;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -24,6 +25,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.jupiter.api.AfterEach;
@@ -60,11 +62,9 @@ public class ReactorNettyTcpStompClientTests {
public void setup(TestInfo testInfo) throws Exception {
logger.debug("Setting up before '" + testInfo.getTestMethod().get().getName() + "'");
@SuppressWarnings("deprecation")
int port = org.springframework.util.SocketUtils.findAvailableTcpPort(61613);
TransportConnector stompConnector = createStompConnector();
this.activeMQBroker = new BrokerService();
this.activeMQBroker.addConnector("stomp://127.0.0.1:" + port);
this.activeMQBroker.addConnector(stompConnector);
this.activeMQBroker.setStartAsync(false);
this.activeMQBroker.setPersistent(false);
this.activeMQBroker.setUseJmx(false);
@@ -75,11 +75,17 @@ public class ReactorNettyTcpStompClientTests {
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.afterPropertiesSet();
this.client = new ReactorNettyTcpStompClient("127.0.0.1", port);
this.client = new ReactorNettyTcpStompClient("127.0.0.1", stompConnector.getServer().getSocketAddress().getPort());
this.client.setMessageConverter(new StringMessageConverter());
this.client.setTaskScheduler(taskScheduler);
}
private TransportConnector createStompConnector() throws Exception {
TransportConnector connector = new TransportConnector();
connector.setUri(new URI("stomp://127.0.0.1:0"));
return connector;
}
@AfterEach
public void shutdown() throws Exception {
try {

19
spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerIntegrationTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.messaging.simp.stomp;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
@@ -26,6 +27,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.jupiter.api.AfterEach;
@@ -70,7 +72,8 @@ public class StompBrokerRelayMessageHandlerIntegrationTests {
private TestEventPublisher eventPublisher;
private int port;
// initial value of zero implies that a random ephemeral port should be used
private int port = 0;
@BeforeEach
@@ -78,7 +81,6 @@ public class StompBrokerRelayMessageHandlerIntegrationTests {
public void setup(TestInfo testInfo) throws Exception {
logger.debug("Setting up before '" + testInfo.getTestMethod().get().getName() + "'");
this.port = org.springframework.util.SocketUtils.findAvailableTcpPort(61613);
this.responseChannel = new ExecutorSubscribableChannel();
this.responseHandler = new TestMessageHandler();
this.responseChannel.subscribe(this.responseHandler);
@@ -88,14 +90,25 @@ public class StompBrokerRelayMessageHandlerIntegrationTests {
}
private void startActiveMQBroker() throws Exception {
TransportConnector stompConnector = createStompConnector(this.port);
this.activeMQBroker = new BrokerService();
this.activeMQBroker.addConnector("stomp://localhost:" + this.port);
this.activeMQBroker.addConnector(stompConnector);
this.activeMQBroker.setStartAsync(false);
this.activeMQBroker.setPersistent(false);
this.activeMQBroker.setUseJmx(false);
this.activeMQBroker.getSystemUsage().getMemoryUsage().setLimit(1024 * 1024 * 5);
this.activeMQBroker.getSystemUsage().getTempUsage().setLimit(1024 * 1024 * 5);
this.activeMQBroker.start();
// Reuse existing ephemeral port on restart (i.e., the next time this method
// is invoked) since it will already be configured in the relay
this.port = (this.port != 0 ? this.port : stompConnector.getServer().getSocketAddress().getPort());
}
private TransportConnector createStompConnector(int port) throws Exception {
TransportConnector connector = new TransportConnector();
connector.setUri(new URI("stomp://localhost:" + port));
return connector;
}
private void createAndStartRelay() throws InterruptedException {