Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Detect clients fail over - that haven't removed themselves from 

a brokee

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@634589 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2008-03-07 09:06:30 +00:00
parent 3f9d6e2ef8
commit 1a01d13731
2 changed files with 19 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
View File

@ -79,7 +79,7 @@ public class AdvisoryBroker extends BrokerFilter {
public Subscription addConsumer(ConnectionContext context, ConsumerInfo info) throws Exception { public Subscription addConsumer(ConnectionContext context, ConsumerInfo info) throws Exception {
Subscription answer = next.addConsumer(context, info); Subscription answer = next.addConsumer(context, info);
// Don't advise advisory topics. // Don't advise advisory topics.
if (!AdvisorySupport.isAdvisoryTopic(info.getDestination())) { if (!AdvisorySupport.isAdvisoryTopic(info.getDestination())) {
ActiveMQTopic topic = AdvisorySupport.getConsumerAdvisoryTopic(info.getDestination()); ActiveMQTopic topic = AdvisorySupport.getConsumerAdvisoryTopic(info.getDestination());

22
activemq-core/src/main/java/org/apache/activemq/broker/cluster/ConnectionSplitBroker.java
View File

@ -16,6 +16,7 @@
*/ */
package org.apache.activemq.broker.cluster; package org.apache.activemq.broker.cluster;
import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
@ -31,6 +32,8 @@ import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.broker.Broker; import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerFilter; import org.apache.activemq.broker.BrokerFilter;
import org.apache.activemq.broker.ConnectionContext; import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.TransportConnection;
import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.command.ActiveMQMessage; import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.BrokerId; import org.apache.activemq.command.BrokerId;
import org.apache.activemq.command.ConnectionId; import org.apache.activemq.command.ConnectionId;
@ -50,14 +53,14 @@ import org.apache.commons.logging.LogFactory;
public class ConnectionSplitBroker extends BrokerFilter implements MessageListener{ public class ConnectionSplitBroker extends BrokerFilter implements MessageListener{
private static final Log LOG = LogFactory.getLog(ConnectionSplitBroker.class); private static final Log LOG = LogFactory.getLog(ConnectionSplitBroker.class);
private Connection connection; private Connection connection;
private Map <ConnectionId,ConnectionInfo>clientMap = new ConcurrentHashMap<ConnectionId,ConnectionInfo>(); private Map <ConnectionId,ConnectionContext>clientMap = new ConcurrentHashMap<ConnectionId,ConnectionContext>();
public ConnectionSplitBroker(Broker next) { public ConnectionSplitBroker(Broker next) {
super(next); super(next);
} }
public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception { public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
if (info != null){ if (info != null){
clientMap.put(info.getConnectionId(),info); clientMap.put(info.getConnectionId(),context);
} }
super.addConnection(context, info); super.addConnection(context, info);
} }
@ -77,6 +80,7 @@ public class ConnectionSplitBroker extends BrokerFilter implements MessageListen
fac.setWarnAboutUnstartedConnectionTimeout(10000); fac.setWarnAboutUnstartedConnectionTimeout(10000);
fac.setWatchTopicAdvisories(false); fac.setWatchTopicAdvisories(false);
fac.setAlwaysSessionAsync(true); fac.setAlwaysSessionAsync(true);
fac.setClientID(getBrokerId().toString()+":" + getBrokerName() + ":ConnectionSplitBroker");
connection = fac.createConnection(); connection = fac.createConnection();
connection.start(); connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
@ -101,8 +105,18 @@ public class ConnectionSplitBroker extends BrokerFilter implements MessageListen
String brokerId=null; String brokerId=null;
try { try {
brokerId = message.getStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID); brokerId = message.getStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID);
if (brokerId != null && brokerId.equals(getBrokerId().getValue())) { 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 {
old.getConnection().stop();
} catch (Exception e) {
LOG.error("Failed to remove stale connection: " + str);
}
}
} }
} catch (JMSException e) { } catch (JMSException e) {
LOG.warn("Failed to get message property "+ AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID,e); LOG.warn("Failed to get message property "+ AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID,e);