AMQ-6858 - Allow configuration of the client Id token in network bridge The client id token that is used to separate parts of a generated local client id in a network bridge should be configurable so bridge names and broker names can contain underscores if desired
(cherry picked from commit 3ca439cadaa8c40f506aa1ef683b36624d216254) Project: http://git-wip-us.apache.org/repos/asf/activemq/repo Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/39d6321a Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/39d6321a Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/39d6321a Branch: refs/heads/activemq-5.15.x Commit: 39d6321a4a2e9f929ba9cc451cafa8f244e3d716 Parents: 97fe20a Author: Christopher L. Shannon (cshannon) <christopher.l.shan...@gmail.com> Authored: Tue Nov 14 14:26:22 2017 -0500 Committer: Christopher L. Shannon (cshannon) <christopher.l.shan...@gmail.com> Committed: Thu Nov 16 07:50:31 2017 -0500 ---------------------------------------------------------------------- .../network/DemandForwardingBridgeSupport.java | 25 +- .../network/NetworkBridgeConfiguration.java | 9 + .../DurableFiveBrokerNetworkBridgeTest.java | 293 ++++++++++--------- 3 files changed, 167 insertions(+), 160 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/activemq/blob/39d6321a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java ---------------------------------------------------------------------- diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java index 03e79e4..75084d1 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java +++ b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java @@ -37,6 +37,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.regex.Pattern; import javax.management.ObjectName; @@ -492,7 +493,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br localConnectionInfo = new ConnectionInfo(); localConnectionInfo.setConnectionId(new ConnectionId(idGenerator.generateId())); - localClientId = configuration.getName() + "_" + remoteBrokerName + "_inbound_" + configuration.getBrokerName(); + localClientId = configuration.getName() + configuration.getClientIdToken() + remoteBrokerName + configuration.getClientIdToken() + "inbound" + configuration.getClientIdToken() + configuration.getBrokerName(); localConnectionInfo.setClientId(localClientId); localConnectionInfo.setUserName(configuration.getUserName()); localConnectionInfo.setPassword(configuration.getPassword()); @@ -520,8 +521,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br ConnectionInfo duplexLocalConnectionInfo = new ConnectionInfo(); duplexLocalConnectionInfo.setConnectionId(new ConnectionId(idGenerator.generateId())); - duplexLocalConnectionInfo.setClientId(configuration.getName() + "_" + remoteBrokerName + "_inbound_duplex_" - + configuration.getBrokerName()); + duplexLocalConnectionInfo.setClientId(configuration.getName() + configuration.getClientIdToken() + remoteBrokerName + configuration.getClientIdToken() + "inbound" + configuration.getClientIdToken() + "duplex" + + configuration.getClientIdToken() + configuration.getBrokerName()); duplexLocalConnectionInfo.setUserName(configuration.getUserName()); duplexLocalConnectionInfo.setPassword(configuration.getPassword()); @@ -609,7 +610,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br } remoteConnectionInfo = new ConnectionInfo(); remoteConnectionInfo.setConnectionId(new ConnectionId(idGenerator.generateId())); - remoteConnectionInfo.setClientId(configuration.getName() + "_" + configuration.getBrokerName() + "_outbound"); + remoteConnectionInfo.setClientId(configuration.getName() + configuration.getClientIdToken() + configuration.getBrokerName() + configuration.getClientIdToken() + "outbound"); remoteConnectionInfo.setUserName(configuration.getUserName()); remoteConnectionInfo.setPassword(configuration.getPassword()); remoteBroker.oneway(remoteConnectionInfo); @@ -685,7 +686,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br if (sub != null && path.length > 1 && subName != null) { String b1 = path[path.length-1].toString(); String b2 = path[path.length-2].toString(); - final SubscriptionInfo newSubInfo = new SubscriptionInfo(b2 + "_inbound_" + b1, subName); + final SubscriptionInfo newSubInfo = new SubscriptionInfo(b2 + configuration.getClientIdToken() + "inbound" + configuration.getClientIdToken() + b1, subName); sub.getDurableRemoteSubs().add(newSubInfo); sub.getNetworkDemandConsumerMap().computeIfAbsent(newSubInfo, v -> new AtomicInteger()).incrementAndGet(); LOG.debug("Adding proxy network subscription {} to demand subscription", newSubInfo); @@ -695,15 +696,11 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br } private String getProxyBridgeClientId(SubscriptionInfo info) { - String[] clientIdTokens = info.getClientId().split("_"); - String newClientId = ""; - if (clientIdTokens.length > 2) { - for (int j = clientIdTokens.length - 3; j < clientIdTokens.length; j++) { - newClientId += clientIdTokens[j]; - if (j < clientIdTokens.length -1) { - newClientId += "_"; - } - } + String newClientId = info.getClientId(); + String[] clientIdTokens = newClientId != null ? newClientId.split(Pattern.quote(configuration.getClientIdToken())) : null; + if (clientIdTokens != null && clientIdTokens.length > 2) { + newClientId = clientIdTokens[clientIdTokens.length - 3] + configuration.getClientIdToken() + "inbound" + + configuration.getClientIdToken() + clientIdTokens[clientIdTokens.length -1]; } return newClientId; } http://git-wip-us.apache.org/repos/asf/activemq/blob/39d6321a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java ---------------------------------------------------------------------- diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java index b2ca78a..3c64758 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java +++ b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java @@ -59,6 +59,7 @@ public class NetworkBridgeConfiguration { private String password; private String destinationFilter = null; private String name = "NC"; + private String clientIdToken = "_"; protected List<ActiveMQDestination> excludedDestinations = new CopyOnWriteArrayList<ActiveMQDestination>(); protected List<ActiveMQDestination> dynamicallyIncludedDestinations = new CopyOnWriteArrayList<ActiveMQDestination>(); @@ -191,6 +192,14 @@ public class NetworkBridgeConfiguration { this.brokerName = brokerName; } + public String getClientIdToken() { + return clientIdToken; + } + + public void setClientIdToken(String clientIdToken) { + this.clientIdToken = clientIdToken; + } + /** * @return the networkTTL */ http://git-wip-us.apache.org/repos/asf/activemq/blob/39d6321a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableFiveBrokerNetworkBridgeTest.java ---------------------------------------------------------------------- diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableFiveBrokerNetworkBridgeTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableFiveBrokerNetworkBridgeTest.java index 4a63553..94d7393 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableFiveBrokerNetworkBridgeTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableFiveBrokerNetworkBridgeTest.java @@ -57,6 +57,7 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu connector.setConduitSubscriptions(true); connector.setSyncDurableSubs(true); connector.setNetworkTTL(-1); + connector.setClientIdToken("|"); return connector; } @@ -71,15 +72,15 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu } /** - * BrokerA -> BrokerB -> BrokerC + * Broker_A_A -> Broker_B_B -> Broker_C_C */ protected void testDurablePropagation() throws Exception { // Setup broker networks - bridgeBrokers("BrokerA", "BrokerB"); - bridgeBrokers("BrokerB", "BrokerC"); + bridgeBrokers("Broker_A_A", "Broker_B_B"); + bridgeBrokers("Broker_B_B", "Broker_C_C"); if (!duplex) { - bridgeBrokers("BrokerB", "BrokerA"); - bridgeBrokers("BrokerC", "BrokerB"); + bridgeBrokers("Broker_B_B", "Broker_A_A"); + bridgeBrokers("Broker_C_C", "Broker_B_B"); } startAllBrokers(); @@ -88,26 +89,26 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu ActiveMQTopic dest = (ActiveMQTopic) createDestination("TEST.FOO", true); // Setup consumers - Session ses = createSession("BrokerA"); + Session ses = createSession("Broker_A_A"); MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA"); MessageConsumer clientB = ses.createDurableSubscriber(dest, "subB"); // let consumers propagate around the network - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); - sendMessages("BrokerC", dest, 1); + sendMessages("Broker_C_C", dest, 1); assertNotNull(clientA.receive(1000)); assertNotNull(clientB.receive(1000)); //bring online a consumer on the other side - Session ses2 = createSession("BrokerC"); + Session ses2 = createSession("Broker_C_C"); MessageConsumer clientC = ses2.createDurableSubscriber(dest, "subC"); //there will be 2 network durables, 1 for each direction of the bridge - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 2); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1); clientA.close(); clientB.close(); @@ -116,9 +117,9 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu ses.unsubscribe("subB"); ses2.unsubscribe("subC"); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0); } @@ -134,11 +135,11 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu protected void testDurablePropagationConsumerAllBrokers() throws Exception { // Setup broker networks - bridgeBrokers("BrokerA", "BrokerB"); - bridgeBrokers("BrokerB", "BrokerC"); + bridgeBrokers("Broker_A_A", "Broker_B_B"); + bridgeBrokers("Broker_B_B", "Broker_C_C"); if (!duplex) { - bridgeBrokers("BrokerB", "BrokerA"); - bridgeBrokers("BrokerC", "BrokerB"); + bridgeBrokers("Broker_B_B", "Broker_A_A"); + bridgeBrokers("Broker_C_C", "Broker_B_B"); } startAllBrokers(); @@ -147,28 +148,28 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu ActiveMQTopic dest = (ActiveMQTopic) createDestination("TEST.FOO", true); // Setup consumers - Session ses = createSession("BrokerA"); + Session ses = createSession("Broker_A_A"); MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA"); // let consumers propagate around the network - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); //bring online a consumer on the other side - Session ses2 = createSession("BrokerB"); + Session ses2 = createSession("Broker_B_B"); MessageConsumer clientB = ses2.createDurableSubscriber(dest, "subB"); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1); - Session ses3 = createSession("BrokerC"); + Session ses3 = createSession("Broker_C_C"); MessageConsumer clientC = ses3.createDurableSubscriber(dest, "subC"); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 2); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1); clientA.close(); @@ -179,9 +180,9 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu ses3.unsubscribe("subC"); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0); } @@ -197,15 +198,15 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu protected void testDurablePropagation5Broker() throws Exception { // Setup broker networks - bridgeBrokers("BrokerA", "BrokerB"); - bridgeBrokers("BrokerB", "BrokerC"); - bridgeBrokers("BrokerC", "BrokerD"); - bridgeBrokers("BrokerD", "BrokerE"); + bridgeBrokers("Broker_A_A", "Broker_B_B"); + bridgeBrokers("Broker_B_B", "Broker_C_C"); + bridgeBrokers("Broker_C_C", "Broker_D_D"); + bridgeBrokers("Broker_D_D", "Broker_E_E"); if (!duplex) { - bridgeBrokers("BrokerB", "BrokerA"); - bridgeBrokers("BrokerC", "BrokerB"); - bridgeBrokers("BrokerD", "BrokerC"); - bridgeBrokers("BrokerE", "BrokerD"); + bridgeBrokers("Broker_B_B", "Broker_A_A"); + bridgeBrokers("Broker_C_C", "Broker_B_B"); + bridgeBrokers("Broker_D_D", "Broker_C_C"); + bridgeBrokers("Broker_E_E", "Broker_D_D"); } startAllBrokers(); @@ -214,42 +215,42 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu ActiveMQTopic dest = (ActiveMQTopic) createDestination("TEST.FOO", true); // Setup consumers - Session ses = createSession("BrokerA"); + Session ses = createSession("Broker_A_A"); MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA"); Thread.sleep(1000); // let consumers propagate around the network - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerD").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerE").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); - sendMessages("BrokerE", dest, 1); + sendMessages("Broker_E_E", dest, 1); assertNotNull(clientA.receive(1000)); //bring online a consumer on the other side - Session ses2 = createSession("BrokerE"); + Session ses2 = createSession("Broker_E_E"); MessageConsumer clientE = ses2.createDurableSubscriber(dest, "subE"); Thread.sleep(1000); //there will be 2 network durables, 1 for each direction of the bridge - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 2); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 2); - assertNCDurableSubsCount(brokers.get("BrokerD").broker, dest, 2); - assertNCDurableSubsCount(brokers.get("BrokerE").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 2); + assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 2); + assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1); clientA.close(); clientE.close(); ses.unsubscribe("subA"); ses2.unsubscribe("subE"); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerD").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerE").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 0); } @@ -265,13 +266,13 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu protected void testDurablePropagationSpoke() throws Exception { // Setup broker networks - bridgeBrokers("BrokerA", "BrokerB"); - bridgeBrokers("BrokerB", "BrokerC"); - bridgeBrokers("BrokerB", "BrokerD"); + bridgeBrokers("Broker_A_A", "Broker_B_B"); + bridgeBrokers("Broker_B_B", "Broker_C_C"); + bridgeBrokers("Broker_B_B", "Broker_D_D"); if (!duplex) { - bridgeBrokers("BrokerB", "BrokerA"); - bridgeBrokers("BrokerC", "BrokerB"); - bridgeBrokers("BrokerD", "BrokerB"); + bridgeBrokers("Broker_B_B", "Broker_A_A"); + bridgeBrokers("Broker_C_C", "Broker_B_B"); + bridgeBrokers("Broker_D_D", "Broker_B_B"); } startAllBrokers(); @@ -280,42 +281,42 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu ActiveMQTopic dest = (ActiveMQTopic) createDestination("TEST.FOO", true); // Setup consumers - Session ses = createSession("BrokerA"); - Session ses2 = createSession("BrokerB"); - Session ses3 = createSession("BrokerC"); - Session ses4 = createSession("BrokerD"); + Session ses = createSession("Broker_A_A"); + Session ses2 = createSession("Broker_B_B"); + Session ses3 = createSession("Broker_C_C"); + Session ses4 = createSession("Broker_D_D"); MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA"); MessageConsumer clientAB = ses.createDurableSubscriber(dest, "subAB"); Thread.sleep(1000); // let consumers propagate around the network - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerD").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); MessageConsumer clientD = ses4.createDurableSubscriber(dest, "subD"); Thread.sleep(1000); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 2); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerD").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1); - sendMessages("BrokerA", dest, 1); + sendMessages("Broker_A_A", dest, 1); assertNotNull(clientD.receive(1000)); - sendMessages("BrokerC", dest, 1); + sendMessages("Broker_C_C", dest, 1); assertNotNull(clientD.receive(1000)); MessageConsumer clientB = ses2.createDurableSubscriber(dest, "subB"); MessageConsumer clientC = ses3.createDurableSubscriber(dest, "subC"); Thread.sleep(1000); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 3); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerD").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 3); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1); clientA.close(); clientAB.close(); @@ -329,10 +330,10 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu ses3.unsubscribe("subC"); ses4.unsubscribe("subD"); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerD").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 0); } public void testForceDurablePropagationDuplex() throws Exception { @@ -347,11 +348,11 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu protected void testForceDurablePropagation() throws Exception { // Setup broker networks - bridgeBrokers("BrokerA", "BrokerB"); - bridgeBrokers("BrokerB", "BrokerC"); + bridgeBrokers("Broker_A_A", "Broker_B_B"); + bridgeBrokers("Broker_B_B", "Broker_C_C"); if (!duplex) { - bridgeBrokers("BrokerB", "BrokerA"); - bridgeBrokers("BrokerC", "BrokerB"); + bridgeBrokers("Broker_B_B", "Broker_A_A"); + bridgeBrokers("Broker_C_C", "Broker_B_B"); } startAllBrokers(); @@ -360,32 +361,32 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu ActiveMQTopic dest = (ActiveMQTopic) createDestination("TEST.FOO", true); // Setup consumers - Session ses = createSession("BrokerA"); + Session ses = createSession("Broker_A_A"); MessageConsumer clientA = ses.createConsumer(dest); Thread.sleep(1000); // let consumers propagate around the network - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); - sendMessages("BrokerC", dest, 1); + sendMessages("Broker_C_C", dest, 1); assertNotNull(clientA.receive(1000)); - Session ses2 = createSession("BrokerC"); + Session ses2 = createSession("Broker_C_C"); MessageConsumer clientC = ses2.createConsumer(dest); Thread.sleep(1000); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 2); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1); clientA.close(); clientC.close(); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0); } public void testDurablePropagationSyncDuplex() throws Exception { @@ -400,14 +401,14 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu protected void testDurablePropagationSync() throws Exception { // Setup broker networks - NetworkConnector nc1 = bridgeBrokers("BrokerA", "BrokerB"); - NetworkConnector nc2 = bridgeBrokers("BrokerB", "BrokerC"); + NetworkConnector nc1 = bridgeBrokers("Broker_A_A", "Broker_B_B"); + NetworkConnector nc2 = bridgeBrokers("Broker_B_B", "Broker_C_C"); NetworkConnector nc3 = null; NetworkConnector nc4 = null; if (!duplex) { - nc3 = bridgeBrokers("BrokerB", "BrokerA"); - nc4 = bridgeBrokers("BrokerC", "BrokerB"); + nc3 = bridgeBrokers("Broker_B_B", "Broker_A_A"); + nc4 = bridgeBrokers("Broker_C_C", "Broker_B_B"); } startAllBrokers(); @@ -424,16 +425,16 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu ActiveMQTopic dest = (ActiveMQTopic) createDestination("TEST.FOO", true); // Setup consumers - Session ses = createSession("BrokerA"); - Session ses2 = createSession("BrokerC"); + Session ses = createSession("Broker_A_A"); + Session ses2 = createSession("Broker_C_C"); MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA"); MessageConsumer clientB = ses.createDurableSubscriber(dest, "subB"); MessageConsumer clientC = ses2.createDurableSubscriber(dest, "subC"); Thread.sleep(1000); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0); nc1.start(); nc2.start(); @@ -443,9 +444,9 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu } //there will be 2 network durables, 1 for each direction of the bridge - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 2); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1); clientA.close(); clientB.close(); @@ -456,14 +457,14 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu duplex = true; // Setup broker networks - bridgeBrokers("BrokerA", "BrokerB"); - bridgeBrokers("BrokerB", "BrokerC"); + bridgeBrokers("Broker_A_A", "Broker_B_B"); + bridgeBrokers("Broker_B_B", "Broker_C_C"); //Duplicate the bridges with different included destinations - valid use case - NetworkConnector nc3 = bridgeBrokers("BrokerA", "BrokerB"); - NetworkConnector nc4 = bridgeBrokers("BrokerB", "BrokerC"); - nc3.setName("nc3"); - nc4.setName("nc4"); + NetworkConnector nc3 = bridgeBrokers("Broker_A_A", "Broker_B_B"); + NetworkConnector nc4 = bridgeBrokers("Broker_B_B", "Broker_C_C"); + nc3.setName("nc_3_3"); + nc4.setName("nc_4_4"); nc3.setDynamicallyIncludedDestinations( Lists.<ActiveMQDestination> newArrayList(new ActiveMQTopic("TEST.FOO2?forceDurable=true"))); nc4.setDynamicallyIncludedDestinations( @@ -476,8 +477,8 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu ActiveMQTopic dest2 = (ActiveMQTopic) createDestination("TEST.FOO2", true); // Setup consumers - Session ses = createSession("BrokerA"); - Session ses2 = createSession("BrokerC"); + Session ses = createSession("Broker_A_A"); + Session ses2 = createSession("Broker_C_C"); MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA"); MessageConsumer clientAa = ses.createDurableSubscriber(dest2, "subAa"); MessageConsumer clientC = ses2.createDurableSubscriber(dest, "subC"); @@ -485,33 +486,33 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu Thread.sleep(1000); //make sure network durables are online - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 2); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 1); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest2, 2); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest2, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest2, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest2, 2); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest2, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest2, 1); clientA.close(); clientC.close(); ses.unsubscribe("subA"); ses2.unsubscribe("subC"); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest, 0); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest2, 2); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest2, 1); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest2, 1); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest2, 2); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest2, 1); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest2, 1); clientAa.close(); clientCc.close(); ses.unsubscribe("subAa"); ses2.unsubscribe("subCc"); - assertNCDurableSubsCount(brokers.get("BrokerB").broker, dest2, 0); - assertNCDurableSubsCount(brokers.get("BrokerC").broker, dest2, 0); - assertNCDurableSubsCount(brokers.get("BrokerA").broker, dest2, 0); + assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest2, 0); + assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest2, 0); + assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest2, 0); } protected void assertNCDurableSubsCount(final BrokerService brokerService, final ActiveMQTopic dest, @@ -552,11 +553,11 @@ public class DurableFiveBrokerNetworkBridgeTest extends JmsMultipleBrokersTestSu super.setAutoFail(true); super.setUp(); String options = new String("?persistent=false&useJmx=false"); - createBroker(new URI("broker:(tcp://localhost:61616)/BrokerA" + options)); - createBroker(new URI("broker:(tcp://localhost:61617)/BrokerB" + options)); - createBroker(new URI("broker:(tcp://localhost:61618)/BrokerC" + options)); - createBroker(new URI("broker:(tcp://localhost:61619)/BrokerD" + options)); - createBroker(new URI("broker:(tcp://localhost:61620)/BrokerE" + options)); + createBroker(new URI("broker:(tcp://localhost:61616)/Broker_A_A" + options)); + createBroker(new URI("broker:(tcp://localhost:61617)/Broker_B_B" + options)); + createBroker(new URI("broker:(tcp://localhost:61618)/Broker_C_C" + options)); + createBroker(new URI("broker:(tcp://localhost:61619)/Broker_D_D" + options)); + createBroker(new URI("broker:(tcp://localhost:61620)/Broker_E_E" + options)); } @Override
