Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix STOMP connect failure related memory leak

Browse Source 
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
This commit is contained in:
Rossen Stoyanchev
2016-05-12 11:21:41 -04:00
parent 6eba220d1b
commit b5022108c7
2 changed files with 19 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
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()));
}

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

@@ -168,7 +168,7 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
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 class Reactor2TcpClient<P> implements TcpOperations<P> {
promise.onError(new Consumer<Throwable>() {
@Override
public void accept(Throwable ex) {
cleanupTask.run();
connectionHandler.afterConnectFailure(ex);
}
})