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

https://issues.apache.org/jira/browse/AMQ-4082 - fix regression with org.apache.activemq.usecases.NetworkOfTwentyBrokersTest nd org.apache.activemq.usecases.RequestReplyNoAdvisoryNetworkTest - statically included dests in duplex case being ignored and unregister without register 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1442122 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2013-02-04 14:06:06 +00:00
parent 856e5061d7
commit c372448471
1 changed files with 50 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
View File

@ -261,6 +261,15 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
l.onStop(this); l.onStop(this);
} }
try { try {
// local start complete
if (startedLatch.getCount() < 2) {
if (LOG.isTraceEnabled()) {
LOG.trace(configuration.getBrokerName() + " unregister bridge (" + this + ") to " + remoteBrokerName);
}
brokerService.getBroker().removeBroker(null, remoteBrokerInfo);
brokerService.getBroker().networkBridgeStopped(remoteBrokerInfo);
}
remoteBridgeStarted.set(false); remoteBridgeStarted.set(false);
final CountDownLatch sendShutdown = new CountDownLatch(1); final CountDownLatch sendShutdown = new CountDownLatch(1);
@ -306,15 +315,11 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
} }
} }
if (remoteBrokerInfo != null) {
brokerService.getBroker().removeBroker(null, remoteBrokerInfo);
brokerService.getBroker().networkBridgeStopped(remoteBrokerInfo);
if (LOG.isInfoEnabled()) { if (LOG.isInfoEnabled()) {
LOG.info(configuration.getBrokerName() + " bridge to " + remoteBrokerName + " stopped"); LOG.info(configuration.getBrokerName() + " bridge to " + remoteBrokerName + " stopped");
} }
} }
} }
}
protected void triggerStartAsyncNetworkBridgeCreation() throws IOException { protected void triggerStartAsyncNetworkBridgeCreation() throws IOException {
brokerService.getTaskRunnerFactory().execute(new Runnable() { brokerService.getTaskRunnerFactory().execute(new Runnable() {
@ -381,21 +386,16 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
} }
private void doStartLocalAndRemoteBridges() { private void doStartLocalAndRemoteBridges() {
try {
startLocalBridge();
} catch (Throwable e) {
serviceLocalException(e);
return;
}
try {
if (disposed.get()) { if (disposed.get()) {
return; return;
} }
Properties props = MarshallingSupport.stringToProperties(remoteBrokerInfo.getNetworkProperties()); if (isCreatedByDuplex()) {
// apply remote (propagated) configuration to local duplex bridge before start
Properties props = null;
try { try {
props = MarshallingSupport.stringToProperties(remoteBrokerInfo.getNetworkProperties());
IntrospectionSupport.getProperties(configuration, props, null); IntrospectionSupport.getProperties(configuration, props, null);
if (configuration.getExcludedDestinations() != null) { if (configuration.getExcludedDestinations() != null) {
excludedDestinations = configuration.getExcludedDestinations().toArray( excludedDestinations = configuration.getExcludedDestinations().toArray(
@ -410,14 +410,19 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
new ActiveMQDestination[configuration.getDynamicallyIncludedDestinations().size()]); new ActiveMQDestination[configuration.getDynamicallyIncludedDestinations().size()]);
} }
} catch (Throwable t) { } catch (Throwable t) {
LOG.error("Error mapping remote destinations", t); LOG.error("Error mapping remote configuration: " + props, t);
}
} }
// Let the local broker know the remote broker's ID. try {
localBroker.oneway(remoteBrokerInfo); startLocalBridge();
// new peer broker (a consumer can work with remote broker also) } catch (Throwable e) {
brokerService.getBroker().addBroker(null, remoteBrokerInfo); serviceLocalException(e);
return;
}
try {
startRemoteBridge();
startRemoteBridge(); startRemoteBridge();
} catch (Throwable e) { } catch (Throwable e) {
serviceRemoteException(e); serviceRemoteException(e);
@ -454,7 +459,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
localBroker.oneway(localSessionInfo); localBroker.oneway(localSessionInfo);
if (configuration.isDuplex()) { if (configuration.isDuplex()) {
// separate in-bound chamnel for forwards so we don't // separate in-bound channel for forwards so we don't
// contend with out-bound dispatch on same connection // contend with out-bound dispatch on same connection
ConnectionInfo duplexLocalConnectionInfo = new ConnectionInfo(); ConnectionInfo duplexLocalConnectionInfo = new ConnectionInfo();
duplexLocalConnectionInfo.setConnectionId(new ConnectionId(idGenerator.generateId())); duplexLocalConnectionInfo.setConnectionId(new ConnectionId(idGenerator.generateId()));
@ -483,10 +488,17 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
l.onStart(this); l.onStart(this);
} }
// Let the local broker know the remote broker's ID.
localBroker.oneway(remoteBrokerInfo);
// new peer broker (a consumer can work with remote broker also)
brokerService.getBroker().addBroker(null, remoteBrokerInfo);
if (LOG.isInfoEnabled()) { if (LOG.isInfoEnabled()) {
LOG.info("Network connection between " + localBroker + " and " + remoteBroker + "(" + remoteBrokerName + ") has been established."); LOG.info("Network connection between " + localBroker + " and " + remoteBroker + "(" + remoteBrokerName + ") has been established.");
if (LOG.isTraceEnabled()) {
LOG.trace(configuration.getBrokerName() + " register bridge (" + this + ") to " + remoteBrokerName);
}
} }
} else { } else {
LOG.warn("Bridge was disposed before the startLocalBridge() method was fully executed."); LOG.warn("Bridge was disposed before the startLocalBridge() method was fully executed.");
} }
@ -1137,7 +1149,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
LOG.error("Failed to add static destination " + dest, e); LOG.error("Failed to add static destination " + dest, e);
} }
if (LOG.isTraceEnabled()) { if (LOG.isTraceEnabled()) {
LOG.trace("bridging messages for static destination: " + dest); LOG.trace(configuration.getBrokerName() + ", bridging messages for static destination: " + dest);
} }
} }
} }