From 7f29df4d795ed76465f2dcd36ed1879c0887ab0e Mon Sep 17 00:00:00 2001 From: Sam Brannen <104798+sbrannen@users.noreply.github.com> Date: Tue, 10 Feb 2026 13:58:24 +0100 Subject: [PATCH] Make StompBrokerRelayMessageHandler's isPauseable() logic dynamic Prior to this commit, an attempt to restart the StompBrokerRelayMessageHandler (SBRMH) resulted in an IllegalStateException stating that the ReactorNettyTcpClient was still in the process of "Shutting down." The reason is that a ReactorNettyTcpClient cannot be restarted after it has been closed, and that is by design. To address that issue, this commit introduces an `internallyManagedTcpClient` flag in SBRMH that is used to track whether SBRMH is in charge of managing the TcpClient internally or if the TcpClient was supplied by the user and is therefore managed externally. If SBRMH manages the TcpClient internally, isPauseable() returns `true`, and the `tcpClient` field is set to null (in stopInternal()) when the handler is stopped. Consequently, a new internally managed TcpClient will be created when the handler is restarted. If the TcpClient is managed externally, the handler is not considered to be "pauseable", and a reference to the externally managed TcpClient is retained after the handler has been stopped. If the ApplicationContext is subsequently restarted, the externally managed TcpClient will be reused -- which may or may not work, depending on the implementation of the TcpClient. Note, however, that this has always been the behavior of SBRMH with regard to stop/start scenarios for externally managed TcpClients. Closes gh-36266 --- .../stomp/StompBrokerRelayMessageHandler.java | 30 ++++- ...tractStompBrokerRelayIntegrationTests.java | 105 +++++++++++++++--- ...NettyStompBrokerRelayIntegrationTests.java | 16 +++ 3 files changed, 137 insertions(+), 14 deletions(-) diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java index 1c6514ab3cd..cab92b50ac8 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java @@ -76,6 +76,7 @@ import org.springframework.util.Assert; * * @author Rossen Stoyanchev * @author Andy Wilkinson + * @author Sam Brannen * @since 4.0 */ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler { @@ -131,6 +132,15 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler private @Nullable TcpOperations tcpClient; + /** + * Tracks whether this {@code StompBrokerRelayMessageHandler} manages the + * TCP client internally. + * @since 7.0.4 + * @see #setTcpClient(TcpOperations) + * @see #isPauseable() + */ + private boolean internallyManagedTcpClient = true; + private @Nullable MessageHeaderInitializer headerInitializer; private final DefaultStats stats = new DefaultStats(); @@ -344,10 +354,12 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler *

By default {@link ReactorNettyTcpClient} is used. *

Note: when this property is used, any * {@link #setRelayHost(String) host} or {@link #setRelayPort(int) port} - * specified are effectively ignored. + * specified will be effectively ignored, and {@link #isPauseable()} will + * return {@code false}. */ public void setTcpClient(@Nullable TcpOperations tcpClient) { this.tcpClient = tcpClient; + this.internallyManagedTcpClient = (tcpClient == null); } /** @@ -415,6 +427,19 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler return this.taskScheduler; } + /** + * Returns {@code true} if this {@code StompBrokerRelayMessageHandler} manages + * the TCP client internally. + *

Returns {@code false} if an externally managed TCP client has been + * {@linkplain #setTcpClient(TcpOperations) registered}. + * @since 7.0.4 + * @see #setTcpClient(TcpOperations) + * @see org.springframework.context.SmartLifecycle#isPauseable() + */ + @Override + public boolean isPauseable() { + return this.internallyManagedTcpClient; + } @Override protected void startInternal() { @@ -472,6 +497,9 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler catch (Throwable ex) { logger.error("Error in shutdown of TCP client", ex); } + if (this.internallyManagedTcpClient) { + this.tcpClient = null; + } } } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/AbstractStompBrokerRelayIntegrationTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/AbstractStompBrokerRelayIntegrationTests.java index 2777b83ca83..135b1c9cbf0 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/AbstractStompBrokerRelayIntegrationTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/AbstractStompBrokerRelayIntegrationTests.java @@ -20,6 +20,7 @@ import java.net.URI; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; @@ -40,10 +41,12 @@ import org.junit.jupiter.api.extension.RegisterExtension; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.StubMessageChannel; +import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.simp.SimpMessageHeaderAccessor; import org.springframework.messaging.simp.SimpMessageType; import org.springframework.messaging.simp.broker.BrokerAvailabilityEvent; @@ -59,6 +62,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType; * Integration tests for {@link StompBrokerRelayMessageHandler} running against ActiveMQ. * * @author Rossen Stoyanchev + * @author Sam Brannen */ public abstract class AbstractStompBrokerRelayIntegrationTests { @@ -119,22 +123,22 @@ public abstract class AbstractStompBrokerRelayIntegrationTests { private void createAndStartRelay() throws InterruptedException { StubMessageChannel channel = new StubMessageChannel(); - List prefixes = Arrays.asList("/queue/", "/topic/"); - this.relay = new StompBrokerRelayMessageHandler(channel, this.responseChannel, channel, prefixes); - // We do not set the relayPort, since we explicitly set the TCP client. - // this.relay.setRelayPort(this.port); + List prefixes = List.of("/queue/", "/topic/"); + + this.relay = createRelay(channel, this.responseChannel, channel, prefixes, this.port); this.relay.setApplicationEventPublisher(this.eventPublisher); this.relay.setSystemHeartbeatReceiveInterval(0); this.relay.setSystemHeartbeatSendInterval(0); this.relay.setPreservePublishOrder(true); - TcpOperations tcpClient = initTcpClient(this.port); - this.relay.setTcpClient(tcpClient); - this.relay.start(); this.eventPublisher.expectBrokerAvailabilityEvent(true); } + protected abstract StompBrokerRelayMessageHandler createRelay(SubscribableChannel inboundChannel, + MessageChannel outboundChannel, SubscribableChannel brokerChannel, Collection destinationPrefixes, + int port); + protected abstract TcpOperations initTcpClient(int port); @AfterEach @@ -207,12 +211,6 @@ public abstract class AbstractStompBrokerRelayIntegrationTests { this.responseHandler.expectMessages(error); } - @Test - void brokerAvailabilityEventWhenStopped() throws Exception { - stopActiveMqBrokerAndAwait(); - this.eventPublisher.expectBrokerAvailabilityEvent(false); - } - @Test void relayReconnectsIfBrokerComesBackUp() throws Exception { String sess1 = "sess1"; @@ -249,6 +247,87 @@ public abstract class AbstractStompBrokerRelayIntegrationTests { this.responseHandler.expectMessages(disconnect); } + @Test // gh-36266 + void stopAndRestartWithInternallyManagedTcpClient() throws Exception { + assertRelayIsRunning(true); + assertRelayIsPauseable(true); + assertBrokerIsAvailable(true); + + publishSubscribe(); + + this.relay.stop(); + assertRelayIsRunning(false); + assertRelayIsPauseable(true); + assertBrokerIsAvailable(false); + this.eventPublisher.expectBrokerAvailabilityEvent(false); + + this.responseHandler.queue.clear(); + + this.relay.start(); + assertRelayIsRunning(true); + assertRelayIsPauseable(true); + if (!this.relay.isBrokerAvailable()) { + this.eventPublisher.expectBrokerAvailabilityEvent(true); + } + assertBrokerIsAvailable(true); + + publishSubscribe(); + } + + @Test // gh-36266 + void stopAndRestartWithExternallyManagedTcpClient() throws Exception { + this.relay.setTcpClient(initTcpClient(this.port)); + + assertRelayIsRunning(true); + assertRelayIsPauseable(false); + assertBrokerIsAvailable(true); + + publishSubscribe(); + + this.relay.stop(); + assertRelayIsRunning(false); + assertRelayIsPauseable(false); + assertBrokerIsAvailable(false); + this.eventPublisher.expectBrokerAvailabilityEvent(false); + + this.responseHandler.queue.clear(); + + this.relay.start(); + assertRelayIsRunning(true); + assertRelayIsPauseable(false); + // Even though the relay is "running", the broker should not be + // available since the TcpClient is externally managed. In other words, + // this is the expected behavior when the relay is not pauseable. + assertBrokerIsAvailable(false); + } + + private void assertRelayIsRunning(boolean running) { + if (running) { + assertThat(this.relay.isRunning()).as("is running").isTrue(); + } + else { + assertThat(this.relay.isRunning()).as("is running").isFalse(); + } + } + + private void assertRelayIsPauseable(boolean pauseable) { + if (pauseable) { + assertThat(this.relay.isPauseable()).as("is pauseable").isTrue(); + } + else { + assertThat(this.relay.isPauseable()).as("is pauseable").isFalse(); + } + } + + private void assertBrokerIsAvailable(boolean available) { + if (available) { + assertThat(this.relay.isBrokerAvailable()).as("is broker available").isTrue(); + } + else { + assertThat(this.relay.isBrokerAvailable()).as("is broker available").isFalse(); + } + } + private static class TestEventPublisher implements ApplicationEventPublisher { diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/ReactorNettyStompBrokerRelayIntegrationTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/ReactorNettyStompBrokerRelayIntegrationTests.java index 1a2216a2bdc..5ce44cd1f8b 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/ReactorNettyStompBrokerRelayIntegrationTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/ReactorNettyStompBrokerRelayIntegrationTests.java @@ -16,6 +16,10 @@ package org.springframework.messaging.simp.stomp; +import java.util.Collection; + +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.tcp.TcpOperations; import org.springframework.messaging.tcp.reactor.ReactorNettyTcpClient; @@ -24,9 +28,21 @@ import org.springframework.messaging.tcp.reactor.ReactorNettyTcpClient; * ActiveMQ with {@link ReactorNettyTcpClient}. * * @author Rossen Stoyanchev + * @author Sam Brannen */ class ReactorNettyStompBrokerRelayIntegrationTests extends AbstractStompBrokerRelayIntegrationTests { + @Override + protected StompBrokerRelayMessageHandler createRelay(SubscribableChannel inboundChannel, + MessageChannel outboundChannel, SubscribableChannel brokerChannel, Collection destinationPrefixes, + int port) { + + StompBrokerRelayMessageHandler handler = + new StompBrokerRelayMessageHandler(inboundChannel, outboundChannel, brokerChannel, destinationPrefixes); + handler.setRelayPort(port); + return handler; + } + @Override protected TcpOperations initTcpClient(int port) { return new ReactorNettyTcpClient<>("127.0.0.1", port, new StompReactorNettyCodec());