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

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<byte[]> 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
 	 * <p>By default {@link ReactorNettyTcpClient} is used.
 	 * <p><strong>Note:</strong> 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<byte[]> 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.
+	 * <p>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;
+			}
 		}
 	}
 

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<String> 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<String> 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<byte[]> 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<String> destinationPrefixes,
+			int port);
+
 	protected abstract TcpOperations<byte[]> 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 {
 

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<String> destinationPrefixes,
+			int port) {
+
+		StompBrokerRelayMessageHandler handler =
+				new StompBrokerRelayMessageHandler(inboundChannel, outboundChannel, brokerChannel, destinationPrefixes);
+		handler.setRelayPort(port);
+		return handler;
+	}
+
 	@Override
 	protected TcpOperations<byte[]> initTcpClient(int port) {
 		return new ReactorNettyTcpClient<>("127.0.0.1", port, new StompReactorNettyCodec());