mirror of https://github.com/apache/activemq.git
Check for old network Subscriptions as well as connections
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@636420 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies
parent
f67415d879
commit
58cf400836
|
|
@ -34,112 +34,170 @@ import org.apache.activemq.broker.BrokerFilter;
|
|||
import org.apache.activemq.broker.ConnectionContext;
|
||||
import org.apache.activemq.broker.TransportConnection;
|
||||
import org.apache.activemq.broker.TransportConnector;
|
||||
import org.apache.activemq.broker.region.Subscription;
|
||||
import org.apache.activemq.command.ActiveMQMessage;
|
||||
import org.apache.activemq.command.BrokerId;
|
||||
import org.apache.activemq.command.ConnectionId;
|
||||
import org.apache.activemq.command.ConnectionInfo;
|
||||
import org.apache.activemq.command.ConsumerId;
|
||||
import org.apache.activemq.command.ConsumerInfo;
|
||||
import org.apache.activemq.command.DataStructure;
|
||||
import org.apache.activemq.command.Message;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
/**
|
||||
* Monitors for client connections that may fail to another
|
||||
* broker - but this broker isn't aware they've gone.
|
||||
* Can occur with network glitches or client error
|
||||
* Monitors for client connections that may fail to another broker - but this
|
||||
* broker isn't aware they've gone. Can occur with network glitches or client
|
||||
* error
|
||||
*
|
||||
* @version $Revision$
|
||||
*/
|
||||
public class ConnectionSplitBroker extends BrokerFilter implements MessageListener{
|
||||
private static final Log LOG = LogFactory.getLog(ConnectionSplitBroker.class);
|
||||
public class ConnectionSplitBroker extends BrokerFilter implements
|
||||
MessageListener {
|
||||
private static final Log LOG = LogFactory
|
||||
.getLog(ConnectionSplitBroker.class);
|
||||
|
||||
private Connection connection;
|
||||
private Map <ConnectionId,ConnectionContext>clientMap = new ConcurrentHashMap<ConnectionId,ConnectionContext>();
|
||||
|
||||
private Map<ConnectionId, ConnectionContext> clientMap = new ConcurrentHashMap<ConnectionId, ConnectionContext>();
|
||||
private Map<ConsumerId,ConsumerInfo>consumerMap = new ConcurrentHashMap<ConsumerId,ConsumerInfo>();
|
||||
public ConnectionSplitBroker(Broker next) {
|
||||
super(next);
|
||||
}
|
||||
|
||||
public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
|
||||
if (info != null){
|
||||
clientMap.put(info.getConnectionId(),context);
|
||||
}
|
||||
public void addConnection(ConnectionContext context, ConnectionInfo info)
|
||||
throws Exception {
|
||||
if (info != null) {
|
||||
removeStaleConnection(info);
|
||||
clientMap.put(info.getConnectionId(), context);
|
||||
}
|
||||
super.addConnection(context, info);
|
||||
}
|
||||
|
||||
public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error)
|
||||
throws Exception {
|
||||
if (info != null){
|
||||
clientMap.remove(info.getConnectionId());
|
||||
}
|
||||
super.removeConnection(context, info, error);
|
||||
public void removeConnection(ConnectionContext context,
|
||||
ConnectionInfo info, Throwable error) throws Exception {
|
||||
if (info != null) {
|
||||
clientMap.remove(info.getConnectionId());
|
||||
}
|
||||
super.removeConnection(context, info, error);
|
||||
}
|
||||
|
||||
public void start() throws Exception{
|
||||
super.start();
|
||||
ActiveMQConnectionFactory fac = new ActiveMQConnectionFactory(getBrokerService().getVmConnectorURI());
|
||||
fac.setCloseTimeout(1);
|
||||
fac.setWarnAboutUnstartedConnectionTimeout(10000);
|
||||
fac.setWatchTopicAdvisories(false);
|
||||
fac.setAlwaysSessionAsync(true);
|
||||
fac.setClientID(getBrokerId().toString()+":" + getBrokerName() + ":ConnectionSplitBroker");
|
||||
connection = fac.createConnection();
|
||||
connection.start();
|
||||
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
MessageConsumer consumer = session.createConsumer(AdvisorySupport.getConnectionAdvisoryTopic());
|
||||
consumer.setMessageListener(this);
|
||||
}
|
||||
|
||||
public synchronized void stop() throws Exception{
|
||||
if (connection != null){
|
||||
connection.stop();
|
||||
connection = null;
|
||||
}
|
||||
super.stop();
|
||||
public Subscription addConsumer(ConnectionContext context, ConsumerInfo info) throws Exception{
|
||||
|
||||
if (info.isNetworkSubscription()) {
|
||||
List<ConsumerId>list = info.getNetworkConsumerIds();
|
||||
for (ConsumerId id:list) {
|
||||
consumerMap.put(id,info);
|
||||
}
|
||||
}else {
|
||||
ConsumerInfo networkInfo = consumerMap.get(info.getConsumerId());
|
||||
if (networkInfo != null) {
|
||||
networkInfo.removeNetworkConsumerId(info.getConsumerId());
|
||||
if (networkInfo.isNetworkConsumersEmpty()) {
|
||||
consumerMap.remove(info.getConsumerId());
|
||||
super.removeConsumer(context,networkInfo);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
return super.addConsumer(context, info);
|
||||
}
|
||||
|
||||
public void onMessage(javax.jms.Message m) {
|
||||
ActiveMQMessage message = (ActiveMQMessage) m;
|
||||
|
||||
DataStructure o = message.getDataStructure();
|
||||
if (o != null && o.getClass() == ConnectionInfo.class) {
|
||||
ConnectionInfo info = (ConnectionInfo) o;
|
||||
String brokerId=null;
|
||||
try {
|
||||
brokerId = message.getStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID);
|
||||
if (brokerId != null && !brokerId.equals(getBrokerId().getValue())) {
|
||||
//see if it already exits
|
||||
ConnectionContext old = clientMap.remove(info.getConnectionId());
|
||||
if (old != null && old.getConnection() != null) {
|
||||
String str = "connectionId=" + old.getConnectionId() +",clientId="+old.getClientId();
|
||||
LOG.warn("Removing stale connection: " + str);
|
||||
try {
|
||||
//remove connection states
|
||||
TransportConnection connection = (TransportConnection) old.getConnection();
|
||||
connection.processRemoveConnection(old.getConnectionId());
|
||||
connection.stopAsync();
|
||||
} catch (Exception e) {
|
||||
LOG.error("Failed to remove stale connection: " + str,e);
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to get message property "+ AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID,e);
|
||||
}
|
||||
|
||||
|
||||
public void removeConsumer(ConnectionContext context, ConsumerInfo info) throws Exception{
|
||||
if (info.isNetworkSubscription()) {
|
||||
List<ConsumerId>list = info.getNetworkConsumerIds();
|
||||
for (ConsumerId id:list) {
|
||||
consumerMap.remove(id);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
protected boolean contains(BrokerId[] brokerPath, BrokerId brokerId) {
|
||||
if (brokerPath != null) {
|
||||
for (int i = 0; i < brokerPath.length; i++) {
|
||||
if (brokerId.equals(brokerPath[i])) {
|
||||
return true;
|
||||
}
|
||||
super.removeConsumer(context, info);
|
||||
}
|
||||
|
||||
public void start() throws Exception {
|
||||
super.start();
|
||||
ActiveMQConnectionFactory fac = new ActiveMQConnectionFactory(
|
||||
getBrokerService().getVmConnectorURI());
|
||||
fac.setCloseTimeout(1);
|
||||
fac.setWarnAboutUnstartedConnectionTimeout(10000);
|
||||
fac.setWatchTopicAdvisories(false);
|
||||
fac.setAlwaysSessionAsync(true);
|
||||
fac.setClientID(getBrokerId().toString() + ":" + getBrokerName()
|
||||
+ ":ConnectionSplitBroker");
|
||||
connection = fac.createConnection();
|
||||
connection.start();
|
||||
Session session = connection.createSession(false,
|
||||
Session.AUTO_ACKNOWLEDGE);
|
||||
MessageConsumer consumer = session.createConsumer(AdvisorySupport
|
||||
.getConnectionAdvisoryTopic());
|
||||
consumer.setMessageListener(this);
|
||||
}
|
||||
|
||||
public synchronized void stop() throws Exception {
|
||||
if (connection != null) {
|
||||
connection.stop();
|
||||
connection = null;
|
||||
}
|
||||
super.stop();
|
||||
}
|
||||
|
||||
public void onMessage(javax.jms.Message m) {
|
||||
ActiveMQMessage message = (ActiveMQMessage) m;
|
||||
|
||||
DataStructure o = message.getDataStructure();
|
||||
if (o != null && o.getClass() == ConnectionInfo.class) {
|
||||
ConnectionInfo info = (ConnectionInfo) o;
|
||||
|
||||
String brokerId = null;
|
||||
try {
|
||||
brokerId = message
|
||||
.getStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID);
|
||||
if (brokerId != null
|
||||
&& !brokerId.equals(getBrokerId().getValue())) {
|
||||
// see if it already exits
|
||||
removeStaleConnection(info);
|
||||
}
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to get message property "
|
||||
+ AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID, e);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
protected boolean contains(BrokerId[] brokerPath, BrokerId brokerId) {
|
||||
if (brokerPath != null) {
|
||||
for (int i = 0; i < brokerPath.length; i++) {
|
||||
if (brokerId.equals(brokerPath[i])) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
protected void removeStaleConnection(ConnectionInfo info) {
|
||||
// see if it already exits
|
||||
ConnectionContext old = clientMap.remove(info
|
||||
.getConnectionId());
|
||||
if (old != null && old.getConnection() != null) {
|
||||
String str = "connectionId=" + old.getConnectionId()
|
||||
+ ",clientId=" + old.getClientId();
|
||||
LOG.warn("Removing stale connection: " + str);
|
||||
try {
|
||||
// remove connection states
|
||||
TransportConnection connection = (TransportConnection) old
|
||||
.getConnection();
|
||||
connection.processRemoveConnection(old
|
||||
.getConnectionId());
|
||||
connection.stopAsync();
|
||||
} catch (Exception e) {
|
||||
LOG.error("Failed to remove stale connection: "
|
||||
+ str, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue