Fix STOMP connect failure related memory leak

Normally heartbeats keep connections from hanging. However in some
cases a connection may hang before a CONNECTED frame is received
and heartbeats are put in place. This commit adds a change to enforce
a 60 limit on receiving the CONNECTED frame.

Issue: SPR-14266

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java

@@ -88,6 +88,14 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
 
 	private static final Message<byte[]> HEARTBEAT_MESSAGE;
 
+	/**
+	 * A heartbeat is setup once a CONNECTED frame is received which contains
+	 * the heartbeat settings we need. If we don't receive CONNECTED within
+	 * a minute, the connection is closed proactively.
+	 */
+	private static final int MAX_TIME_TO_CONNECTED_FRAME = 60 * 1000;
+
+
 
 	static {
 		EMPTY_TASK.run();
@@ -567,6 +575,15 @@ public void afterConnected(TcpConnection<byte[]> connection) {
 				logger.debug("TCP connection opened in session=" + getSessionId());
 			}
 			this.tcpConnection = connection;
+			this.tcpConnection.onReadInactivity(new Runnable() {
+				@Override
+				public void run() {
+					if (tcpConnection != null && !isStompConnected) {
+						handleTcpConnectionFailure("No CONNECTED frame received in " +
+								MAX_TIME_TO_CONNECTED_FRAME + " ms.", null);
+					}
+				}
+			}, MAX_TIME_TO_CONNECTED_FRAME);
 			connection.send(MessageBuilder.createMessage(EMPTY_PAYLOAD, this.connectHeaders.getMessageHeaders()));
 		}
 

spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpClient.java

@@ -168,7 +168,7 @@ public ListenableFuture<Void> connect(final TcpConnectionHandler<P> connectionHa
 		Assert.notNull(connectionHandler, "TcpConnectionHandler must not be null");
 
 		final TcpClient<Message<P>, Message<P>> tcpClient;
-		Runnable cleanupTask;
+		final Runnable cleanupTask;
 		synchronized (this.tcpClients) {
 			if (this.stopping) {
 				IllegalStateException ex = new IllegalStateException("Shutting down.");
@@ -194,6 +194,7 @@ public void run() {
 				promise.onError(new Consumer<Throwable>() {
 					@Override
 					public void accept(Throwable ex) {
+						cleanupTask.run();
 						connectionHandler.afterConnectFailure(ex);
 					}
 				})