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

Fix issue in STOMP broker relay

Browse Source
This commit is contained in:
Rossen Stoyanchev
2013-07-15 17:45:54 -04:00
parent 5d20b75dc2
commit e0d393eed3
3 changed files with 13 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-messaging/src/main/java/org/springframework/messaging/simp/handler/UserDestinationMessageHandler.java
View File

@@ -180,7 +180,7 @@ public class UserDestinationMessageHandler implements MessageHandler {
}
public String getTargetDestination(String sessionId) {
return (this.targetDestination != null) ? this.targetDestination + "/" + sessionId : null;
return (this.targetDestination != null) ? this.targetDestination + sessionId : null;
}
}

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

@@ -24,7 +24,6 @@ import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -232,7 +231,7 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
}
this.running = false;
try {
this.tcpClient.close().await(5000, TimeUnit.MILLISECONDS);
this.tcpClient.close().await();
}
catch (Throwable t) {
logger.error("Failed to close reactor TCP client", t);
@@ -321,12 +320,13 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
}
else if (SimpMessageType.DISCONNECT.equals(messageType)) {
RelaySession session = this.relaySessions.remove(sessionId);
if (session != null) {
if (session == null) {
if (logger.isTraceEnabled()) {
logger.trace("Session already removed, sessionId=" + sessionId);
}
session.forward(message);
return;
}
session.forward(message);
}
else {
RelaySession session = this.relaySessions.get(sessionId);
@@ -404,16 +404,9 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
}
return;
}
if (StompCommand.ERROR == headers.getStompCommand()) {
if (logger.isDebugEnabled()) {
logger.warn("STOMP ERROR: " + headers.getMessage() + ". Removing session id=" + this.sessionId);
}
relaySessions.remove(this.sessionId);
}
headers.setSessionId(this.sessionId);
message = MessageBuilder.fromMessage(message).copyHeaders(headers.toMap()).build();
sendMessageToClient(message);
}
@@ -455,25 +448,13 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
}
private boolean forwardInternal(Message<?> message, TcpConnection<String, String> connection) {
if (logger.isTraceEnabled()) {
logger.trace("Forwarding message to STOMP broker, message id=" + message.getHeaders().getId());
}
byte[] bytes = stompMessageConverter.fromMessage(message);
connection.send(new String(bytes, Charset.forName("UTF-8")));
try {
if (logger.isTraceEnabled()) {
logger.trace("Forwarding message to STOMP broker, message id=" + message.getHeaders().getId());
}
byte[] bytes = stompMessageConverter.fromMessage(message);
connection.send(new String(bytes, Charset.forName("UTF-8")));
}
catch (Throwable ex) {
logger.error("Forward failed message id=" + message.getHeaders().getId(), ex);
try {
connection.close();
}
catch (Throwable t) {
// ignore
}
sendError(this.sessionId, "Failed to forward message " + message + ": " + ex.getMessage());
return false;
}
// TODO: detect if send fails and send ERROR downstream (except on DISCONNECT)
return true;
}

4
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompWebSocketHandler.java
View File

@@ -244,11 +244,11 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
WebSocketSession session = this.sessions.get(sessionId);
if (session == null) {
// TODO: failed message delivery mechanism
logger.error("Ignoring message, session not found: " + sessionId);
logger.error("Ignoring message, sessionId not found: " + message);
return;
}
if (headers.getSubscriptionId() == null) {
if (StompCommand.MESSAGE.equals(headers.getStompCommand()) && (headers.getSubscriptionId() == null)) {
// TODO: failed message delivery mechanism
logger.error("Ignoring message, no subscriptionId header: " + message);
return;