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

check if a broker held by the TransportConnector is stopped before asigning it for re-use (i.e. VMTransportServer) 

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@372910 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2006-01-27 17:00:26 +00:00
parent 4e5012ca5d
commit 233b604642
7 changed files with 53 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
activemq-core/src/main/java/org/apache/activemq/broker/Broker.java
View File

@ -197,9 +197,13 @@ public interface Broker extends Region, Service {
public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Throwable;
/**
*
* @return true if the broker is running as a slave
*/
public boolean isSlaveBroker();
/**
* @return true if the broker has stopped
*/
public boolean isStopped();
}

4
activemq-core/src/main/java/org/apache/activemq/broker/BrokerFilter.java
View File

@ -183,5 +183,9 @@ public class BrokerFilter implements Broker {
public boolean isSlaveBroker(){
return next.isSlaveBroker();
}
public boolean isStopped(){
return next.isStopped();
}
}

4
activemq-core/src/main/java/org/apache/activemq/broker/EmptyBroker.java
View File

@ -181,5 +181,9 @@ public class EmptyBroker implements Broker{
public boolean isSlaveBroker(){
return false;
}
public boolean isStopped(){
return false;
}
}

4
activemq-core/src/main/java/org/apache/activemq/broker/ErrorBroker.java
View File

@ -180,5 +180,9 @@ public class ErrorBroker implements Broker {
throw new IllegalStateException(this.message);
}
public boolean isStopped(){
return true;
}
}

4
activemq-core/src/main/java/org/apache/activemq/broker/MutableBrokerFilter.java
View File

@ -193,5 +193,9 @@ public class MutableBrokerFilter implements Broker {
public boolean isSlaveBroker(){
return getNext().isSlaveBroker();
}
public boolean isStopped(){
return getNext().isStopped();
}
}

6
activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java
View File

@ -67,6 +67,7 @@ public class RegionBroker implements Broker {
private final Region tempQueueRegion;
private final Region tempTopicRegion;
private BrokerService brokerService;
private boolean stopped = false;
protected final DestinationStatistics destinationStatistics = new DestinationStatistics();
@ -130,6 +131,7 @@ public class RegionBroker implements Broker {
}
public void stop() throws Exception {
stopped = true;
ServiceStopper ss = new ServiceStopper();
ss.stop(queueRegion);
ss.stop(topicRegion);
@ -442,6 +444,10 @@ public class RegionBroker implements Broker {
public boolean isSlaveBroker(){
return brokerService.isSlave();
}
public boolean isStopped(){
return stopped;
}

28
activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransportFactory.java
View File

@ -88,8 +88,9 @@ public class VMTransportFactory extends TransportFactory {
location = new URI("vm://"+host);
}
VMTransportServer server = (VMTransportServer) servers.get(host);
if( server == null ) {
VMTransportServer server = (VMTransportServer) servers.get(host);
//validate the broker is still active
if( !validateBroker(host) || server == null ) {
BrokerService broker = BrokerRegistry.getInstance().lookup(host);
if (broker == null) {
try {
@ -112,6 +113,8 @@ public class VMTransportFactory extends TransportFactory {
connector.start();
connectors.put(host, connector);
}
}else {
}
VMTransport vmtransport = server.connect();
@ -171,4 +174,25 @@ public class VMTransportFactory extends TransportFactory {
this.brokerFactoryHandler = brokerFactoryHandler;
}
private boolean validateBroker(String host){
boolean result=true;
if(brokers.containsKey(host)||servers.containsKey(host)||connectors.containsKey(host)){
//check the broker is still in the BrokerRegistry
TransportConnector connector=(TransportConnector) connectors.get(host);
if(BrokerRegistry.getInstance().lookup(host)==null||(connector!=null&&connector.getBroker().isStopped())){
result=false;
//clean-up
brokers.remove(host);
servers.remove(host);
if(connector!=null){
connectors.remove(host);
if(connector!=null){
ServiceSupport.dispose(connector);
}
}
}
}
return result;
}
}