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

Polish StompBrokerRelayMessageHandler

Browse Source
This commit is contained in:
Rossen Stoyanchev
2013-08-19 21:10:36 -04:00
parent 94fefec0f9
commit 68c0df8305
1 changed files with 149 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -26,6 +26,7 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -60,17 +61,18 @@ import reactor.tuple.Tuple2;
/**
* A {@link MessageHandler} that handles messages by forwarding them to a STOMP broker and
* reversely sends any returned messages from the broker to the provided
* {@link MessageChannel}.
*
* @author Rossen Stoyanchev
* @author Andy Wilkinson
*
* @since 4.0
*/
public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLifecycle, ApplicationEventPublisherAware {
private static final Log logger = LogFactory.getLog(StompBrokerRelayMessageHandler.class);
private static final String STOMP_RELAY_SYSTEM_SESSION_ID = "stompRelaySystemSessionId";
private final MessageChannel messageChannel;
private final String[] destinationPrefixes;
@@ -99,6 +101,7 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
private AtomicBoolean brokerAvailable = new AtomicBoolean(false);
/**
* @param messageChannel the channel to send messages from the STOMP broker to
* @param destinationPrefixes the broker supported destination prefixes; destinations
@@ -111,6 +114,7 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
this.destinationPrefixes = destinationPrefixes.toArray(new String[destinationPrefixes.size()]);
}
/**
* Set the STOMP message broker host.
*/
@@ -207,7 +211,14 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
.codec(new DelimitedCodec<String, String>((byte) 0, true, StandardCodecs.STRING_CODEC))
.connect(this.relayHost, this.relayPort)
.get();
openSystemSession();
if (logger.isDebugEnabled()) {
logger.debug("Initializing \"system\" TCP connection");
}
SystemRelaySession session = new SystemRelaySession();
this.relaySessions.put(session.getId(), session);
session.connect();
this.running = true;
}
}
@@ -218,46 +229,6 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
this.applicationEventPublisher = applicationEventPublisher;
}
/**
* Open a "system" session for sending messages from parts of the application
* not associated with a client STOMP session.
*/
private void openSystemSession() {
RelaySession session = new RelaySession(STOMP_RELAY_SYSTEM_SESSION_ID) {
@Override
protected void sendMessageToClient(Message<?> message) {
// ignore, only used to send messages
// TODO: ERROR frame/reconnect
}
@Override
protected Composable<TcpConnection<String, String>> openConnection() {
return tcpClient.open(new Reconnect() {
@Override
public Tuple2<InetSocketAddress, Long> reconnect(InetSocketAddress currentAddress, int attempt) {
return Tuple.of(currentAddress, 5000L);
}
});
}
};
this.relaySessions.put(STOMP_RELAY_SYSTEM_SESSION_ID, session);
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
headers.setAcceptVersion("1.1,1.2");
headers.setLogin(this.systemLogin);
headers.setPasscode(this.systemPasscode);
headers.setHeartbeat(0,0); // TODO
if (logger.isDebugEnabled()) {
logger.debug("Sending STOMP CONNECT frame to initialize \"system\" TCP connection");
}
Message<?> message = MessageBuilder.withPayloadAndHeaders(new byte[0], headers).build();
session.open(message);
}
@Override
public void stop() {
synchronized (this.lifecycleMonitor) {
@@ -305,7 +276,7 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
}
if (SimpMessageType.MESSAGE.equals(messageType)) {
sessionId = (sessionId == null) ? STOMP_RELAY_SYSTEM_SESSION_ID : sessionId;
sessionId = (sessionId == null) ? SystemRelaySession.ID : sessionId;
headers.setSessionId(sessionId);
command = (command == null) ? StompCommand.SEND : command;
headers.setCommandIfNotSet(command);
@@ -329,11 +300,11 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
}
if (SimpMessageType.CONNECT.equals(messageType)) {
headers.setHeartbeat(0, 0); // TODO: disable for now
headers.setHeartbeat(0, 0);
message = MessageBuilder.withPayloadAndHeaders(message.getPayload(), headers).build();
RelaySession session = new RelaySession(sessionId);
this.relaySessions.put(sessionId, session);
session.open(message);
session.connect(message);
}
else if (SimpMessageType.DISCONNECT.equals(messageType)) {
RelaySession session = this.relaySessions.remove(sessionId);
@@ -375,7 +346,7 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
return false;
}
private void brokerAvailable() {
private void publishBrokerAvailableEvent() {
if ((this.applicationEventPublisher != null) && this.brokerAvailable.compareAndSet(false, true)) {
if (logger.isTraceEnabled()) {
logger.trace("Publishing BrokerAvailabilityEvent (available)");
@@ -384,7 +355,7 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
}
}
private void brokerUnavailable() {
private void publishBrokerUnavailableEvent() {
if ((this.applicationEventPublisher != null) && this.brokerAvailable.compareAndSet(true, false)) {
if (logger.isTraceEnabled()) {
logger.trace("Publishing BrokerAvailabilityEvent (unavailable)");
@@ -400,9 +371,7 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
private final BlockingQueue<Message<?>> messageQueue = new LinkedBlockingQueue<Message<?>>(50);
private volatile TcpConnection<String, String> connection = null;
private volatile boolean isConnected = false;
private volatile StompConnection stompConnection = new StompConnection();
private final Object monitor = new Object();
@@ -412,40 +381,45 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
this.sessionId = sessionId;
}
public void open(final Message<?> message) {
Assert.notNull(message, "message is required");
Composable<TcpConnection<String, String>> connectionComposable = openConnection();
public String getId() {
return this.sessionId;
}
public void connect(final Message<?> connectMessage) {
Assert.notNull(connectMessage, "connectMessage is required");
Composable<TcpConnection<String, String>> connectionComposable = openTcpConnection();
connectionComposable.consume(new Consumer<TcpConnection<String, String>>() {
@Override
public void accept(TcpConnection<String, String> newConnection) {
isConnected = false;
connection = newConnection;
newConnection.in().consume(new Consumer<String>() {
@Override
public void accept(String stompFrame) {
readStompFrame(stompFrame);
}
});
forwardInternal(message);
public void accept(TcpConnection<String, String> connection) {
handleTcpConnection(connection, connectMessage);
}
});
connectionComposable.when(Throwable.class, new Consumer<Throwable>() {
@Override
public void accept(Throwable ex) {
relaySessions.remove(sessionId);
logger.error("Failed to connect to broker", ex);
sendError(sessionId, "Failed to connect to message broker " + ex.toString());
handleTcpClientFailure("Failed to connect to message broker", ex);
}
});
}
protected Composable<TcpConnection<String, String>> openConnection() {
protected Composable<TcpConnection<String, String>> openTcpConnection() {
return tcpClient.open();
}
protected void handleTcpConnection(TcpConnection<String, String> tcpConn, final Message<?> connectMessage) {
this.stompConnection.setTcpConnection(tcpConn);
tcpConn.in().consume(new Consumer<String>() {
@Override
public void accept(String message) {
readStompFrame(message);
}
});
forwardInternal(tcpConn, connectMessage);
}
private void readStompFrame(String stompFrame) {
// heartbeat
@@ -461,8 +435,8 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
if (StompCommand.CONNECTED == headers.getCommand()) {
synchronized(this.monitor) {
this.isConnected = true;
brokerAvailable();
this.stompConnection.setReady();
publishBrokerAvailableEvent();
flushMessages();
}
return;
@@ -473,37 +447,32 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
sendMessageToClient(message);
}
protected void sendMessageToClient(Message<?> message) {
messageChannel.send(message);
private void handleTcpClientFailure(String message, Throwable ex) {
if (logger.isErrorEnabled()) {
logger.error(message + ", sessionId=" + this.sessionId, ex);
}
this.stompConnection.setDisconnected();
sendError(message);
publishBrokerUnavailableEvent();
}
private void sendError(String sessionId, String errorText) {
disconnect();
private void sendError(String errorText) {
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.ERROR);
headers.setSessionId(sessionId);
headers.setSessionId(this.sessionId);
headers.setMessage(errorText);
Message<?> errorMessage = MessageBuilder.withPayloadAndHeaders(new byte[0], headers).build();
sendMessageToClient(errorMessage);
}
private void disconnect() {
this.isConnected = false;
TcpConnection<String, String> localConnection = this.connection;
if (localConnection != null) {
localConnection.close();
this.connection = null;
}
brokerUnavailable();
protected void sendMessageToClient(Message<?> message) {
messageChannel.send(message);
}
public void forward(Message<?> message) {
if (!this.isConnected) {
if (!this.stompConnection.isReady()) {
synchronized(this.monitor) {
if (!this.isConnected) {
if (!this.stompConnection.isReady()) {
this.messageQueue.add(message);
if (logger.isTraceEnabled()) {
logger.trace("Not connected, message queued. Queue size=" + this.messageQueue.size());
@@ -523,10 +492,14 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
}
private boolean forwardInternal(final Message<?> message) {
TcpConnection<String, String> localConnection = this.connection;
if (localConnection == null) {
TcpConnection<String, String> tcpConnection = this.stompConnection.getReadyConnection();
if (tcpConnection == null) {
return false;
}
return forwardInternal(tcpConnection, message);
}
private boolean forwardInternal(TcpConnection<String, String> tcpConnection, final Message<?> message) {
if (logger.isTraceEnabled()) {
logger.trace("Forwarding to STOMP broker, message: " + message);
@@ -536,7 +509,7 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
String payload = new String(bytes, Charset.forName("UTF-8"));
final Deferred<Boolean, Promise<Boolean>> deferred = new DeferredPromiseSpec<Boolean>().get();
localConnection.send(payload, new Consumer<Boolean>() {
tcpConnection.send(payload, new Consumer<Boolean>() {
@Override
public void accept(Boolean success) {
deferred.accept(success);
@@ -547,17 +520,17 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
try {
success = deferred.compose().await();
if (success == null) {
sendError(sessionId, "Timed out waiting for message to be forwarded to the broker");
handleTcpClientFailure("Timed out waiting for message to be forwarded to the broker", null);
}
else if (!success) {
if (StompHeaderAccessor.wrap(message).getCommand() != StompCommand.DISCONNECT) {
sendError(sessionId, "Failed to forward message to the broker");
handleTcpClientFailure("Failed to forward message to the broker", null);
}
}
}
catch (InterruptedException ie) {
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
sendError(sessionId, "Interrupted while forwarding message to the broker");
handleTcpClientFailure("Interrupted while forwarding message to the broker", ex);
}
return (success != null) ? success : false;
}
@@ -572,4 +545,84 @@ public class StompBrokerRelayMessageHandler implements MessageHandler, SmartLife
}
}
}
private static class StompConnection {
private volatile TcpConnection<String, String> connection;
private AtomicReference<TcpConnection<String, String>> readyConnection =
new AtomicReference<TcpConnection<String, String>>();
public void setTcpConnection(TcpConnection<String, String> connection) {
Assert.notNull(connection, "connection must not be null");
this.connection = connection;
}
public TcpConnection<String, String> getReadyConnection() {
return this.readyConnection.get();
}
public void setReady() {
this.readyConnection.set(this.connection);
}
public boolean isReady() {
return (this.readyConnection.get() != null);
}
public void setDisconnected() {
this.readyConnection.set(null);
this.connection = null;
}
@Override
public String toString() {
return "StompConnection [ready=" + isReady() + "]";
}
}
private class SystemRelaySession extends RelaySession {
public static final String ID = "stompRelaySystemSessionId";
public SystemRelaySession() {
super(ID);
}
public void connect() {
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
headers.setAcceptVersion("1.1,1.2");
headers.setLogin(systemLogin);
headers.setPasscode(systemPasscode);
headers.setHeartbeat(0,0);
Message<?> connectMessage = MessageBuilder.withPayloadAndHeaders(new byte[0], headers).build();
super.connect(connectMessage);
}
@Override
protected Composable<TcpConnection<String, String>> openTcpConnection() {
return tcpClient.open(new Reconnect() {
@Override
public Tuple2<InetSocketAddress, Long> reconnect(InetSocketAddress address, int attempt) {
return Tuple.of(address, 5000L);
}
});
}
@Override
protected void sendMessageToClient(Message<?> message) {
StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
if (StompCommand.ERROR.equals(headers.getCommand())) {
if (logger.isErrorEnabled()) {
logger.error("System session received ERROR frame from broker: " + message);
}
}
else {
// Ignore
}
}
}
}