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