You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by de...@apache.org on 2013/04/08 18:58:21 UTC
svn commit: r1465683 - in /activemq/trunk:
activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverPriorityTest.java
Author: dejanb
Date: Mon Apr 8 16:58:21 2013
New Revision: 1465683
URL: http://svn.apache.org/r1465683
Log:
https://issues.apache.org/jira/browse/AMQ-4461 - priority backup with multiple brokers
Modified:
activemq/trunk/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverPriorityTest.java
Modified: activemq/trunk/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java?rev=1465683&r1=1465682&r2=1465683&view=diff
==============================================================================
--- activemq/trunk/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java (original)
+++ activemq/trunk/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java Mon Apr 8 16:58:21 2013
@@ -1169,9 +1169,11 @@ public class FailoverTransport implement
t.setTransportListener(bt);
t.start();
bt.setTransport(t);
- backups.add(bt);
if (priorityBackup && isPriority(uri)) {
priorityBackupAvailable = true;
+ backups.add(0, bt);
+ } else {
+ backups.add(bt);
}
}
} catch (Exception e) {
Modified: activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverPriorityTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverPriorityTest.java?rev=1465683&r1=1465682&r2=1465683&view=diff
==============================================================================
--- activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverPriorityTest.java (original)
+++ activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverPriorityTest.java Mon Apr 8 16:58:21 2013
@@ -16,10 +16,12 @@
*/
package org.apache.activemq.transport.failover;
+import org.apache.activemq.broker.BrokerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
+import java.util.Map;
public class FailoverPriorityTest extends FailoverClusterTestSupport {
@@ -27,6 +29,7 @@ public class FailoverPriorityTest extend
private static final String BROKER_A_CLIENT_TC_ADDRESS = "tcp://127.0.0.1:61616";
private static final String BROKER_B_CLIENT_TC_ADDRESS = "tcp://127.0.0.1:61617";
+ private static final String BROKER_C_CLIENT_TC_ADDRESS = "tcp://127.0.0.1:61618";
private HashMap<String,String> urls = new HashMap<String,String>();
@Override
@@ -38,6 +41,7 @@ public class FailoverPriorityTest extend
private static final String BROKER_A_NAME = "BROKERA";
private static final String BROKER_B_NAME = "BROKERB";
+ private static final String BROKER_C_NAME = "BROKERC";
public void testPriorityBackup() throws Exception {
@@ -86,6 +90,42 @@ public class FailoverPriorityTest extend
restart(false, BROKER_B_NAME, BROKER_A_NAME);
}
+
+ public void testThreeBrokers() throws Exception {
+ // Broker A
+ addBroker(BROKER_A_NAME, createBroker(BROKER_A_NAME));
+ addTransportConnector(getBroker(BROKER_A_NAME), "openwire", BROKER_A_CLIENT_TC_ADDRESS, false);
+ addNetworkBridge(getBroker(BROKER_A_NAME), "A_2_B_Bridge", "static://(" + BROKER_B_CLIENT_TC_ADDRESS + ")?useExponentialBackOff=false", false, null);
+ addNetworkBridge(getBroker(BROKER_A_NAME), "A_2_C_Bridge", "static://(" + BROKER_C_CLIENT_TC_ADDRESS + ")?useExponentialBackOff=false", false, null);
+ getBroker(BROKER_A_NAME).start();
+
+ // Broker B
+ addBroker(BROKER_B_NAME, createBroker(BROKER_B_NAME));
+ addTransportConnector(getBroker(BROKER_B_NAME), "openwire", BROKER_B_CLIENT_TC_ADDRESS, false);
+ addNetworkBridge(getBroker(BROKER_B_NAME), "B_2_A_Bridge", "static://(" + BROKER_A_CLIENT_TC_ADDRESS + ")?useExponentialBackOff=false", false, null);
+ addNetworkBridge(getBroker(BROKER_B_NAME), "B_2_C_Bridge", "static://(" + BROKER_C_CLIENT_TC_ADDRESS + ")?useExponentialBackOff=false", false, null);
+ getBroker(BROKER_B_NAME).start();
+
+ // Broker C
+ addBroker(BROKER_C_NAME, createBroker(BROKER_C_NAME));
+ addTransportConnector(getBroker(BROKER_C_NAME), "openwire", BROKER_C_CLIENT_TC_ADDRESS, false);
+ addNetworkBridge(getBroker(BROKER_C_NAME), "C_2_A_Bridge", "static://(" + BROKER_A_CLIENT_TC_ADDRESS + ")?useExponentialBackOff=false", false, null);
+ addNetworkBridge(getBroker(BROKER_C_NAME), "C_2_B_Bridge", "static://(" + BROKER_B_CLIENT_TC_ADDRESS + ")?useExponentialBackOff=false", false, null);
+ getBroker(BROKER_C_NAME).start();
+
+
+ getBroker(BROKER_C_NAME).waitUntilStarted();
+ Thread.sleep(1000);
+
+ setClientUrl("failover:(" + BROKER_A_CLIENT_TC_ADDRESS + "," + BROKER_B_CLIENT_TC_ADDRESS + "," + BROKER_C_CLIENT_TC_ADDRESS + ")?randomize=false&priorityBackup=true&initialReconnectDelay=1000&useExponentialBackOff=false&backupPoolSize=2");
+
+ createClients(5);
+
+ assertAllConnectedTo(urls.get(BROKER_A_NAME));
+
+ restart(true, BROKER_A_NAME, BROKER_B_NAME);
+
+ }
private void restart(boolean primary, String primaryName, String secondaryName) throws Exception {