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

Applied patch for https://issues.apache.org/activemq/browse/AMQ-1511 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@646245 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2008-04-09 09:45:22 +00:00
parent 60ba951914
commit f1b6362ded
2 changed files with 35 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java
View File

@ -439,12 +439,17 @@ public class BrokerService implements Service {
getBroker().start(); getBroker().start();
/* // see if there is a MasterBroker service and if so, configure
* if(isUseJmx()){ // yes - this is order dependent! // register all // it and start it.
* destination in persistence store including inactive destinations for (Service service : services) {
* as mbeans this.startDestinationsInPersistenceStore(broker); } if (service instanceof MasterConnector) {
*/ configureService(service);
service.start();
}
}
if (!isSlave()) {
startAllConnectors(); startAllConnectors();
}
if (isUseJmx() && masterConnector != null) { if (isUseJmx() && masterConnector != null) {
registerFTConnectorMBean(masterConnector); registerFTConnectorMBean(masterConnector);
@ -1301,6 +1306,7 @@ public class BrokerService implements Service {
* @throws Exception * @throws Exception
*/ */
protected void processHelperProperties() throws Exception { protected void processHelperProperties() throws Exception {
boolean masterServiceExists = false;
if (transportConnectorURIs != null) { if (transportConnectorURIs != null) {
for (int i = 0; i < transportConnectorURIs.length; i++) { for (int i = 0; i < transportConnectorURIs.length; i++) {
String uri = transportConnectorURIs[i]; String uri = transportConnectorURIs[i];
@ -1319,8 +1325,14 @@ public class BrokerService implements Service {
addJmsConnector(jmsBridgeConnectors[i]); addJmsConnector(jmsBridgeConnectors[i]);
} }
} }
for (Service service : services) {
if (service instanceof MasterConnector) {
masterServiceExists = true;
break;
}
}
if (masterConnectorURI != null) { if (masterConnectorURI != null) {
if (masterConnector != null) { if (masterServiceExists) {
throw new IllegalStateException("Cannot specify masterConnectorURI when a masterConnector is already registered via the services property"); throw new IllegalStateException("Cannot specify masterConnectorURI when a masterConnector is already registered via the services property");
} else { } else {
addService(new MasterConnector(masterConnectorURI)); addService(new MasterConnector(masterConnectorURI));
@ -1846,11 +1858,13 @@ public class BrokerService implements Service {
BrokerServiceAware serviceAware = (BrokerServiceAware) service; BrokerServiceAware serviceAware = (BrokerServiceAware) service;
serviceAware.setBrokerService(this); serviceAware.setBrokerService(this);
} }
if (masterConnector == null) {
if (service instanceof MasterConnector) { if (service instanceof MasterConnector) {
masterConnector = (MasterConnector) service; masterConnector = (MasterConnector) service;
supportFailOver = true; supportFailOver = true;
} }
} }
}
/** /**
* Starts all destiantions in persistence store. This includes all inactive * Starts all destiantions in persistence store. This includes all inactive

10
activemq-core/src/main/java/org/apache/activemq/broker/ft/MasterConnector.java
View File

@ -104,7 +104,7 @@ public class MasterConnector implements Service, BrokerServiceAware {
} }
localBroker = TransportFactory.connect(localURI); localBroker = TransportFactory.connect(localURI);
remoteBroker = TransportFactory.connect(remoteURI); remoteBroker = TransportFactory.connect(remoteURI);
LOG.info("Starting a network connection between " + localBroker + " and " + remoteBroker + " has been established."); LOG.info("Starting a slave connection between " + localBroker + " and " + remoteBroker + " has been established.");
localBroker.setTransportListener(new DefaultTransportListener() { localBroker.setTransportListener(new DefaultTransportListener() {
public void onCommand(Object command) { public void onCommand(Object command) {
@ -131,22 +131,16 @@ public class MasterConnector implements Service, BrokerServiceAware {
} }
} }
}); });
masterActive.set(true);
Thread thead = new Thread() {
public void run() {
try { try {
localBroker.start(); localBroker.start();
remoteBroker.start(); remoteBroker.start();
startBridge(); startBridge();
masterActive.set(true);
} catch (Exception e) { } catch (Exception e) {
masterActive.set(false); masterActive.set(false);
LOG.error("Failed to start network bridge: " + e, e); LOG.error("Failed to start network bridge: " + e, e);
} }
} }
};
thead.start();
}
protected void startBridge() throws Exception { protected void startBridge() throws Exception {
connectionInfo = new ConnectionInfo(); connectionInfo = new ConnectionInfo();