Use ConnectionLostException after heartbeat failures

ConnectionLostException was applies only after the WebSocket library
notified us of a session close. However, read inactivity and heartbeat
send failures are also cases of the connection being lost rather than
closed intentionally.

This commit also ensures resetConnection is called after a heartbeat
write failure, consistent with other places where a transport error
is handled that implies the connection is lost.

See gh-32195

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -29,4 +29,8 @@ public class ConnectionLostException extends RuntimeException {
 		super(message);
 	}
 
+	public ConnectionLostException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
 }

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

@@ -714,9 +714,14 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
 		public void run() {
 			TcpConnection<byte[]> conn = connection;
 			if (conn != null) {
-				conn.sendAsync(HEARTBEAT).whenComplete((unused, throwable) -> {
-					if (throwable != null) {
-						handleFailure(throwable);
+				conn.sendAsync(HEARTBEAT).whenComplete((unused, ex) -> {
+					if (ex != null) {
+						String msg = "Heartbeat write failure. Closing connection in session id=" + sessionId + ".";
+						if (logger.isDebugEnabled()) {
+							logger.debug(msg);
+						}
+						resetConnection();
+						handleFailure(new ConnectionLostException(msg, ex));
 					}
 				});
 			}
@@ -728,13 +733,13 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
 
 		@Override
 		public void run() {
-			String error = "Read inactivity. Closing connection in session id=" + sessionId + ".";
+			String msg = "Read inactivity. Closing connection in session id=" + sessionId + ".";
 			if (logger.isDebugEnabled()) {
-				logger.debug(error);
+				logger.debug(msg);
 			}
 			clientSideClose = true;
 			resetConnection();
-			handleFailure(new IllegalStateException(error));
+			handleFailure(new ConnectionLostException(msg));
 		}
 	}
 

spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/DefaultStompSessionTests.java

@@ -221,7 +221,7 @@ public class DefaultStompSessionTests {
 
 		reset(this.sessionHandler);
 		readTask.run();
-		verify(this.sessionHandler).handleTransportError(same(this.session), any(IllegalStateException.class));
+		verify(this.sessionHandler).handleTransportError(same(this.session), any(ConnectionLostException.class));
 		verifyNoMoreInteractions(this.sessionHandler);
 	}