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

Refine BrokerAvailabilityEvent behavior

Browse Source 
Add accessor for brokerAvailable in AbstractBrokerMessageHandler
Ensure brokerAvailable is set even if eventPublisher is not
Add tests BrokerMessageHandlerTests

Turn off brokerAvailable when StompBrokerRelayMessageHandler stops
Actually stop message handling when brokerAvailable=false
Improve log messages

Issue: SPR-11563
This commit is contained in:
Rossen Stoyanchev
2014-03-16 14:16:34 -04:00
parent 14a8f19670
commit 6bcbb94aba
10 changed files with 275 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/AbstractBrokerMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -30,9 +30,10 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.CollectionUtils;
/**
* Abstract base class for a {@link MessageHandler} that manages subscriptions and
* propagates messages to subscribers.
* Abstract base class for a {@link MessageHandler} that broker messages to
* registered subscribers.
*
* @author Rossen Stoyanchev
* @since 4.0
@@ -42,7 +43,7 @@ public abstract class AbstractBrokerMessageHandler
protected final Log logger = LogFactory.getLog(getClass());
private Collection<String> destinationPrefixes;
private final Collection<String> destinationPrefixes;
private ApplicationEventPublisher eventPublisher;
@@ -55,9 +56,13 @@ public abstract class AbstractBrokerMessageHandler
private volatile boolean running = false;
public AbstractBrokerMessageHandler() {
this(Collections.<String>emptyList());
}
public AbstractBrokerMessageHandler(Collection<String> destinationPrefixes) {
this.destinationPrefixes = (destinationPrefixes != null)
? destinationPrefixes : Collections.<String>emptyList();
destinationPrefixes = (destinationPrefixes != null) ? destinationPrefixes : Collections.<String>emptyList();
this.destinationPrefixes = Collections.unmodifiableCollection(destinationPrefixes);
}
@@ -88,6 +93,13 @@ public abstract class AbstractBrokerMessageHandler
return Integer.MAX_VALUE;
}
/**
* Check whether this message handler is currently running.
*
* <p>Note that even when this message handler is running the
* {@link #isBrokerAvailable()} flag may still independently alternate between
* being on and off depending on the concrete sub-class implementation.
*/
@Override
public final boolean isRunning() {
synchronized (this.lifecycleMonitor) {
@@ -95,6 +107,23 @@ public abstract class AbstractBrokerMessageHandler
}
}
/**
* Whether the message broker is currently available and able to process messages.
*
* <p>Note that this is in addition to the {@link #isRunning()} flag, which
* indicates whether this message handler is running. In other words the message
* handler must first be running and then the {@link #isBrokerAvailable()} flag
* may still independently alternate between being on and off depending on the
* concrete sub-class implementation.
*
* <p>Application components may implement
* {@link org.springframework.context.ApplicationListener<BrokerAvailabilityEvent>>}
* to receive notifications when broker becomes available and unavailable.
*/
public boolean isBrokerAvailable() {
return this.brokerAvailable.get();
}
@Override
public final void start() {
synchronized (this.lifecycleMonitor) {
@@ -157,18 +186,20 @@ public abstract class AbstractBrokerMessageHandler
}
protected void publishBrokerAvailableEvent() {
if ((this.eventPublisher != null) && this.brokerAvailable.compareAndSet(false, true)) {
if (logger.isTraceEnabled()) {
logger.trace("Publishing BrokerAvailabilityEvent (available)");
boolean shouldPublish = this.brokerAvailable.compareAndSet(false, true);
if (this.eventPublisher != null && shouldPublish) {
if (logger.isDebugEnabled()) {
logger.debug("Publishing BrokerAvailabilityEvent (available)");
}
this.eventPublisher.publishEvent(new BrokerAvailabilityEvent(true, this));
}
}
protected void publishBrokerUnavailableEvent() {
if ((this.eventPublisher != null) && this.brokerAvailable.compareAndSet(true, false)) {
if (logger.isTraceEnabled()) {
logger.trace("Publishing BrokerAvailabilityEvent (unavailable)");
boolean shouldPublish = this.brokerAvailable.compareAndSet(true, false);
if (this.eventPublisher != null && shouldPublish) {
if (logger.isDebugEnabled()) {
logger.debug("Publishing BrokerAvailabilityEvent (unavailable)");
}
this.eventPublisher.publishEvent(new BrokerAvailabilityEvent(false, this));
}

2
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/SimpleBrokerMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.

2
spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractMessageBrokerConfiguration.java
View File

@@ -355,7 +355,7 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
}
private static final AbstractBrokerMessageHandler noopBroker = new AbstractBrokerMessageHandler(null) {
private static final AbstractBrokerMessageHandler noopBroker = new AbstractBrokerMessageHandler() {
@Override
protected void startInternal() {

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

@@ -126,7 +126,6 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
Assert.notNull(clientOutChannel, "'clientOutChannel' must not be null");
Assert.notNull(brokerChannel, "'brokerChannel' must not be null");
this.clientInboundChannel = clientInChannel;
this.clientOutboundChannel = clientOutChannel;
this.brokerChannel = brokerChannel;
@@ -328,7 +327,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
}
if (logger.isDebugEnabled()) {
logger.debug("Initializing \"system\" TCP connection");
logger.debug("Initializing \"system\" connection");
}
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
@@ -347,6 +346,8 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
@Override
protected void stopInternal() {
publishBrokerUnavailableEvent();
this.clientInboundChannel.unsubscribe(this);
this.brokerChannel.unsubscribe(this);
@@ -358,6 +359,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
logger.error("Failed to close connection in session " + handler.getSessionId() + ": " + t.getMessage());
}
}
try {
this.tcpClient.shutdown();
}
@@ -371,6 +373,17 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
String sessionId = headers.getSessionId();
if (!isBrokerAvailable()) {
if (sessionId == null || sessionId == SystemStompConnectionHandler.SESSION_ID) {
throw new MessageDeliveryException("Message broker is not active.");
}
if (logger.isTraceEnabled()) {
logger.trace("Message broker is not active. Ignoring message id=" + message.getHeaders().getId());
}
return;
}
String destination = headers.getDestination();
StompCommand command = headers.getCommand();
SimpMessageType messageType = headers.getMessageType();
@@ -396,9 +409,14 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
return;
}
if (logger.isTraceEnabled()) {
logger.trace("Processing message=" + message);
}
if (SimpMessageType.CONNECT.equals(messageType)) {
logger.debug("Processing CONNECT in session=" + sessionId +
", number of connections=" + this.connectionHandlers.size());
if (logger.isDebugEnabled()) {
logger.debug("Processing CONNECT (total connected=" + this.connectionHandlers.size() + ")");
}
headers.setLogin(this.clientLogin);
headers.setPasscode(this.clientPasscode);
if (getVirtualHost() != null) {
@@ -412,7 +430,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
StompConnectionHandler handler = this.connectionHandlers.get(sessionId);
if (handler == null) {
if (logger.isTraceEnabled()) {
logger.trace("Connection already removed for sessionId=" + sessionId);
logger.trace("Connection already removed for sessionId '" + sessionId + "'");
}
return;
}
@@ -422,7 +440,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
StompConnectionHandler handler = this.connectionHandlers.get(sessionId);
if (handler == null) {
if (logger.isWarnEnabled()) {
logger.warn("Connection for sessionId=" + sessionId + " not found. Ignoring message");
logger.warn("Connection for sessionId '" + sessionId + "' not found. Ignoring message");
}
return;
}
@@ -466,7 +484,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
@Override
public void afterConnected(TcpConnection<byte[]> connection) {
if (logger.isDebugEnabled()) {
logger.debug("Established TCP connection to broker in session=" + this.sessionId);
logger.debug("Established TCP connection to broker in session '" + this.sessionId + "'");
}
this.tcpConnection = connection;
connection.send(MessageBuilder.withPayload(EMPTY_PAYLOAD).setHeaders(this.connectHeaders).build());
@@ -483,7 +501,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
*/
protected void handleTcpConnectionFailure(String errorMessage, Throwable ex) {
if (logger.isErrorEnabled()) {
logger.error(errorMessage + ", sessionId=" + this.sessionId, ex);
logger.error(errorMessage + ", sessionId '" + this.sessionId + "'", ex);
}
try {
sendStompErrorToClient(errorMessage);
@@ -524,7 +542,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
logger.trace("Received broker heartbeat");
}
else if (logger.isDebugEnabled()) {
logger.debug("Received broker message in session=" + this.sessionId);
logger.debug("Received message from broker in session '" + this.sessionId + "'");
}
if (StompCommand.CONNECTED == headers.getCommand()) {
@@ -595,6 +613,9 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
return;
}
try {
if (logger.isDebugEnabled()) {
logger.debug("TCP connection to broker closed in session '" + this.sessionId + "'");
}
sendStompErrorToClient("Connection to broker closed");
}
finally {
@@ -678,6 +699,10 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
}
finally {
if (this.isRemoteClientSession) {
if (logger.isDebugEnabled()) {
logger.debug("Removing session '" + sessionId + "' (total remaining=" +
(StompBrokerRelayMessageHandler.this.connectionHandlers.size() - 1) + ")");
}
StompBrokerRelayMessageHandler.this.connectionHandlers.remove(this.sessionId);
}
}