diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java index 59e81cdf84..f2329856e3 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java @@ -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((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> openConnection() { - return tcpClient.open(new Reconnect() { - @Override - public Tuple2 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> messageQueue = new LinkedBlockingQueue>(50); - private volatile TcpConnection 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> connectionComposable = openConnection(); + public String getId() { + return this.sessionId; + } + public void connect(final Message connectMessage) { + Assert.notNull(connectMessage, "connectMessage is required"); + + Composable> connectionComposable = openTcpConnection(); connectionComposable.consume(new Consumer>() { @Override - public void accept(TcpConnection newConnection) { - isConnected = false; - connection = newConnection; - newConnection.in().consume(new Consumer() { - @Override - public void accept(String stompFrame) { - readStompFrame(stompFrame); - } - }); - forwardInternal(message); + public void accept(TcpConnection connection) { + handleTcpConnection(connection, connectMessage); } }); - connectionComposable.when(Throwable.class, new Consumer() { @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> openConnection() { + protected Composable> openTcpConnection() { return tcpClient.open(); } + protected void handleTcpConnection(TcpConnection tcpConn, final Message connectMessage) { + this.stompConnection.setTcpConnection(tcpConn); + tcpConn.in().consume(new Consumer() { + @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 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 localConnection = this.connection; - if (localConnection == null) { + TcpConnection tcpConnection = this.stompConnection.getReadyConnection(); + if (tcpConnection == null) { return false; } + return forwardInternal(tcpConnection, message); + } + + private boolean forwardInternal(TcpConnection 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> deferred = new DeferredPromiseSpec().get(); - localConnection.send(payload, new Consumer() { + tcpConnection.send(payload, new Consumer() { @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 connection; + + private AtomicReference> readyConnection = + new AtomicReference>(); + + + public void setTcpConnection(TcpConnection connection) { + Assert.notNull(connection, "connection must not be null"); + this.connection = connection; + } + + public TcpConnection 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> openTcpConnection() { + return tcpClient.open(new Reconnect() { + @Override + public Tuple2 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 + } + } + } + }