You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by an...@apache.org on 2014/11/11 12:00:40 UTC
[10/51] [partial] activemq-6 git commit: ACTIVEMQ6-2 Update to HQ
master
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java
index 5540517..97d45f6 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java
@@ -47,6 +47,12 @@ import org.hornetq.core.client.impl.Topology;
import org.hornetq.core.client.impl.TopologyMemberImpl;
import org.hornetq.core.config.ClusterConnectionConfiguration;
import org.hornetq.core.config.Configuration;
+import org.hornetq.core.config.HAPolicyConfiguration;
+import org.hornetq.core.config.ha.LiveOnlyPolicyConfiguration;
+import org.hornetq.core.config.ha.ReplicaPolicyConfiguration;
+import org.hornetq.core.config.ha.ReplicatedPolicyConfiguration;
+import org.hornetq.core.config.ha.SharedStoreMasterPolicyConfiguration;
+import org.hornetq.core.config.ha.SharedStoreSlavePolicyConfiguration;
import org.hornetq.core.postoffice.Binding;
import org.hornetq.core.postoffice.Bindings;
import org.hornetq.core.postoffice.PostOffice;
@@ -58,8 +64,8 @@ import org.hornetq.core.server.HornetQServers;
import org.hornetq.core.server.NodeManager;
import org.hornetq.core.server.cluster.ClusterConnection;
import org.hornetq.core.server.cluster.ClusterManager;
+import org.hornetq.core.server.cluster.HornetQServerSideProtocolManagerFactory;
import org.hornetq.core.server.cluster.RemoteQueueBinding;
-import org.hornetq.core.server.cluster.ha.HAPolicy;
import org.hornetq.core.server.cluster.impl.ClusterConnectionImpl;
import org.hornetq.core.server.cluster.qourum.SharedNothingBackupQuorum;
import org.hornetq.core.server.group.GroupingHandler;
@@ -860,19 +866,19 @@ public abstract class ClusterTestBase extends ServiceTestBase
protected void setUpGroupHandler(final GroupingHandlerConfiguration.TYPE type, final int node, final int timeout)
{
- setUpGroupHandler(type, node, timeout, -1, GroupingHandlerConfiguration.DEFAULT_REAPER_PERIOD);
+ setUpGroupHandler(type, node, timeout, -1, HornetQDefaultConfiguration.getDefaultGroupingHandlerReaperPeriod());
}
protected void setUpGroupHandler(final GroupingHandlerConfiguration.TYPE type, final int node, final int timeout, final long groupTimeout,
final long reaperPeriod)
{
- servers[node].getConfiguration()
- .setGroupingHandlerConfiguration(new GroupingHandlerConfiguration(new SimpleString("grouparbitrator"),
- type,
- new SimpleString("queues"),
- timeout,
- groupTimeout,
- reaperPeriod));
+ servers[node].getConfiguration().setGroupingHandlerConfiguration(new GroupingHandlerConfiguration()
+ .setName(new SimpleString("grouparbitrator"))
+ .setType(type)
+ .setAddress(new SimpleString("queues"))
+ .setTimeout(timeout)
+ .setGroupTimeout(groupTimeout)
+ .setReaperPeriod(reaperPeriod));
}
protected void setUpGroupHandler(final GroupingHandler groupingHandler, final int node)
@@ -1551,6 +1557,8 @@ public abstract class ClusterTestBase extends ServiceTestBase
locators[node] = HornetQClient.createServerLocatorWithoutHA(serverTotc);
}
+ locators[node].setProtocolManagerFactory(HornetQServerSideProtocolManagerFactory.getInstance());
+
locators[node].setBlockOnNonDurableSend(true);
locators[node].setBlockOnDurableSend(true);
addServerLocator(locators[node]);
@@ -1632,33 +1640,44 @@ public abstract class ClusterTestBase extends ServiceTestBase
protected void setupServer(final int node, final boolean fileStorage, final boolean netty) throws Exception
{
- setupLiveServer(node, fileStorage, false, netty);
+ setupLiveServer(node, fileStorage, false, netty, false);
+ }
+
+ protected void setupLiveServer(final int node, final boolean fileStorage, final boolean netty, boolean isLive) throws Exception
+ {
+ setupLiveServer(node, fileStorage, false, netty, isLive);
}
protected void setupLiveServer(final int node,
final boolean fileStorage,
final boolean sharedStorage,
- final boolean netty) throws Exception
+ final boolean netty,
+ boolean liveOnly) throws Exception
{
if (servers[node] != null)
{
throw new IllegalArgumentException("Already a server at node " + node);
}
- Configuration configuration = createBasicConfig(node);
-
- configuration.setJournalMaxIO_AIO(1000);
-
- if (sharedStorage)
- configuration.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.SHARED_STORE);
+ HAPolicyConfiguration haPolicyConfiguration = null;
+ if (liveOnly)
+ {
+ haPolicyConfiguration = new LiveOnlyPolicyConfiguration();
+ }
else
- configuration.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.REPLICATED);
-
- configuration.setThreadPoolMaxSize(10);
+ {
+ if (sharedStorage)
+ haPolicyConfiguration = new SharedStoreMasterPolicyConfiguration();
+ else
+ haPolicyConfiguration = new ReplicatedPolicyConfiguration();
+ }
- configuration.getAcceptorConfigurations().clear();
- configuration.getAcceptorConfigurations().add(createTransportConfiguration(netty, true,
- generateParams(node, netty)));
+ Configuration configuration = createBasicConfig(node)
+ .setJournalMaxIO_AIO(1000)
+ .setThreadPoolMaxSize(10)
+ .clearAcceptorConfigurations()
+ .addAcceptorConfiguration(createTransportConfiguration(netty, true, generateParams(node, netty)))
+ .setHAPolicyConfiguration(haPolicyConfiguration);
HornetQServer server;
@@ -1715,23 +1734,16 @@ public abstract class ClusterTestBase extends ServiceTestBase
throw new IllegalArgumentException("Already a server at node " + node);
}
- Configuration configuration = createBasicConfig(sharedStorage ? liveNode : node);
-
- if (sharedStorage)
- configuration.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_SHARED_STORE);
- else
- configuration.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_REPLICATED);
-
- configuration.getAcceptorConfigurations().clear();
-
- TransportConfiguration acceptorConfig = createTransportConfiguration(netty, true, generateParams(node, netty));
- configuration.getAcceptorConfigurations().add(acceptorConfig);
-
- // add backup connector
TransportConfiguration liveConfig = createTransportConfiguration(netty, false, generateParams(liveNode, netty));
- configuration.getConnectorConfigurations().put(liveConfig.getName(), liveConfig);
TransportConfiguration backupConfig = createTransportConfiguration(netty, false, generateParams(node, netty));
- configuration.getConnectorConfigurations().put(backupConfig.getName(), backupConfig);
+ TransportConfiguration acceptorConfig = createTransportConfiguration(netty, true, generateParams(node, netty));
+
+ Configuration configuration = createBasicConfig(sharedStorage ? liveNode : node)
+ .clearAcceptorConfigurations()
+ .addAcceptorConfiguration(acceptorConfig)
+ .addConnectorConfiguration(liveConfig.getName(), liveConfig)
+ .addConnectorConfiguration(backupConfig.getName(), backupConfig)
+ .setHAPolicyConfiguration(sharedStorage ? new SharedStoreSlavePolicyConfiguration() : new ReplicaPolicyConfiguration());
HornetQServer server;
@@ -1760,41 +1772,35 @@ public abstract class ClusterTestBase extends ServiceTestBase
throw new IllegalArgumentException("Already a server at node " + node);
}
- Configuration configuration = createBasicConfig(node);
-
- configuration.setJournalMaxIO_AIO(1000);
-
- if (sharedStorage)
- configuration.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.SHARED_STORE);
- else
- configuration.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.REPLICATED);
-
- configuration.getAcceptorConfigurations().clear();
-
Map<String, Object> params = generateParams(node, netty);
- configuration.getAcceptorConfigurations().add(createTransportConfiguration(netty, true, params));
-
TransportConfiguration connector = createTransportConfiguration(netty, false, params);
- configuration.getConnectorConfigurations().put(connector.getName(), connector);
List<String> connectorPairs = new ArrayList<String>();
connectorPairs.add(connector.getName());
- UDPBroadcastGroupConfiguration endpoint = new UDPBroadcastGroupConfiguration(groupAddress, port, null, -1);
-
- BroadcastGroupConfiguration bcConfig = new BroadcastGroupConfiguration("bg1",
- 200,
- connectorPairs,
- endpoint);
-
- configuration.getBroadcastGroupConfigurations().add(bcConfig);
-
- DiscoveryGroupConfiguration dcConfig = new DiscoveryGroupConfiguration("dg1",
- 1000,
- 1000, endpoint);
-
- configuration.getDiscoveryGroupConfigurations().put(dcConfig.getName(), dcConfig);
+ UDPBroadcastGroupConfiguration endpoint = new UDPBroadcastGroupConfiguration().setGroupAddress(groupAddress).setGroupPort(port);
+
+ BroadcastGroupConfiguration bcConfig = new BroadcastGroupConfiguration()
+ .setName("bg1")
+ .setBroadcastPeriod(200)
+ .setConnectorInfos(connectorPairs)
+ .setEndpointFactoryConfiguration(endpoint);
+
+ DiscoveryGroupConfiguration dcConfig = new DiscoveryGroupConfiguration()
+ .setName("dg1")
+ .setRefreshTimeout(1000)
+ .setDiscoveryInitialWaitTimeout(1000)
+ .setBroadcastEndpointFactoryConfiguration(endpoint);
+
+ Configuration configuration = createBasicConfig(node)
+ .setJournalMaxIO_AIO(1000)
+ .clearAcceptorConfigurations()
+ .addAcceptorConfiguration(createTransportConfiguration(netty, true, params))
+ .addConnectorConfiguration(connector.getName(), connector)
+ .addBroadcastGroupConfiguration(bcConfig)
+ .addDiscoveryGroupConfiguration(dcConfig.getName(), dcConfig)
+ .setHAPolicyConfiguration(sharedStorage ? new SharedStoreMasterPolicyConfiguration() : new ReplicatedPolicyConfiguration());
HornetQServer server;
if (fileStorage)
@@ -1837,40 +1843,34 @@ public abstract class ClusterTestBase extends ServiceTestBase
throw new IllegalArgumentException("Already a server at node " + node);
}
- Configuration configuration = createBasicConfig(sharedStorage ? liveNode : node);
-
- if (sharedStorage)
- configuration.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_SHARED_STORE);
- else
- configuration.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_REPLICATED);
-
- configuration.getAcceptorConfigurations().clear();
-
Map<String, Object> params = generateParams(node, netty);
- configuration.getAcceptorConfigurations().add(createTransportConfiguration(netty, true, params));
-
TransportConfiguration connector = createTransportConfiguration(netty, false, params);
- configuration.getConnectorConfigurations().put(connector.getName(), connector);
List<String> connectorPairs = new ArrayList<String>();
connectorPairs.add(connector.getName());
- UDPBroadcastGroupConfiguration endpoint = new UDPBroadcastGroupConfiguration(groupAddress, port, null, -1);
-
- BroadcastGroupConfiguration bcConfig = new BroadcastGroupConfiguration("bg1",
- 1000,
- connectorPairs,
- endpoint);
+ UDPBroadcastGroupConfiguration endpoint = new UDPBroadcastGroupConfiguration().setGroupAddress(groupAddress).setGroupPort(port);
- configuration.getBroadcastGroupConfigurations().add(bcConfig);
+ BroadcastGroupConfiguration bcConfig = new BroadcastGroupConfiguration()
+ .setName("bg1")
+ .setBroadcastPeriod(1000)
+ .setConnectorInfos(connectorPairs)
+ .setEndpointFactoryConfiguration(endpoint);
- DiscoveryGroupConfiguration dcConfig = new DiscoveryGroupConfiguration("dg1",
- 5000,
- 5000,
- endpoint);
+ DiscoveryGroupConfiguration dcConfig = new DiscoveryGroupConfiguration()
+ .setName("dg1")
+ .setRefreshTimeout(5000)
+ .setDiscoveryInitialWaitTimeout(5000)
+ .setBroadcastEndpointFactoryConfiguration(endpoint);
- configuration.getDiscoveryGroupConfigurations().put(dcConfig.getName(), dcConfig);
+ Configuration configuration = createBasicConfig(sharedStorage ? liveNode : node)
+ .clearAcceptorConfigurations()
+ .addAcceptorConfiguration(createTransportConfiguration(netty, true, params))
+ .addConnectorConfiguration(connector.getName(), connector)
+ .addBroadcastGroupConfiguration(bcConfig)
+ .addDiscoveryGroupConfiguration(dcConfig.getName(), dcConfig)
+ .setHAPolicyConfiguration(sharedStorage ? new SharedStoreSlavePolicyConfiguration() : new ReplicatedPolicyConfiguration());
HornetQServer server;
if (sharedStorage)
@@ -1935,15 +1935,18 @@ public abstract class ClusterTestBase extends ServiceTestBase
pairs.add(serverTotc.getName());
}
Configuration config = serverFrom.getConfiguration();
- ClusterConnectionConfiguration clusterConf =
- new ClusterConnectionConfiguration(name, address, name,
- 100,
- true,
- forwardWhenNoConsumers,
- maxHops,
- 1024,
- pairs,
- allowDirectConnectionsOnly);
+
+ ClusterConnectionConfiguration clusterConf = new ClusterConnectionConfiguration()
+ .setName(name)
+ .setAddress(address)
+ .setConnectorName(name)
+ .setRetryInterval(100)
+ .setForwardWhenNoConsumers(forwardWhenNoConsumers)
+ .setMaxHops(maxHops)
+ .setConfirmationWindowSize(1024)
+ .setStaticConnectors(pairs)
+ .setAllowDirectConnectionsOnly(allowDirectConnectionsOnly);
+
config.getClusterConfigurations().add(clusterConf);
}
@@ -1978,16 +1981,19 @@ public abstract class ClusterTestBase extends ServiceTestBase
pairs.add(serverTotc.getName());
}
Configuration config = serverFrom.getConfiguration();
- ClusterConnectionConfiguration clusterConf =
- new ClusterConnectionConfiguration(name, address, name,
- retryInterval,
- true,
- forwardWhenNoConsumers,
- maxHops,
- 1024,
- pairs,
- allowDirectConnectionsOnly);
- clusterConf.setReconnectAttempts(reconnectAttempts);
+
+ ClusterConnectionConfiguration clusterConf = new ClusterConnectionConfiguration()
+ .setName(name)
+ .setAddress(address)
+ .setConnectorName(name)
+ .setReconnectAttempts(reconnectAttempts)
+ .setRetryInterval(retryInterval)
+ .setForwardWhenNoConsumers(forwardWhenNoConsumers)
+ .setMaxHops(maxHops)
+ .setConfirmationWindowSize(1024)
+ .setStaticConnectors(pairs)
+ .setAllowDirectConnectionsOnly(allowDirectConnectionsOnly);
+
config.getClusterConfigurations().add(clusterConf);
}
@@ -2054,18 +2060,19 @@ public abstract class ClusterTestBase extends ServiceTestBase
pairs.add(serverTotc.getName());
}
Configuration conf = serverFrom.getConfiguration();
- ClusterConnectionConfiguration clusterConf = new ClusterConnectionConfiguration(
- name, address, connectorFrom.getName(),
- HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
- HornetQDefaultConfiguration.getDefaultClusterFailureCheckPeriod(),
- HornetQDefaultConfiguration.getDefaultClusterConnectionTtl(),
- retryInterval,
- HornetQDefaultConfiguration.getDefaultClusterRetryIntervalMultiplier(),
- HornetQDefaultConfiguration.getDefaultClusterMaxRetryInterval(),
- -1, reconnectAttempts, 1000, 1000, true, forwardWhenNoConsumers, maxHops,
- 1024, pairs, false,
- HornetQDefaultConfiguration.getDefaultClusterNotificationInterval(),
- HornetQDefaultConfiguration.getDefaultClusterNotificationAttempts(), null);
+
+ ClusterConnectionConfiguration clusterConf = new ClusterConnectionConfiguration()
+ .setName(name)
+ .setAddress(address)
+ .setConnectorName(connectorFrom.getName())
+ .setRetryInterval(retryInterval)
+ .setReconnectAttempts(reconnectAttempts)
+ .setCallTimeout(100)
+ .setCallFailoverTimeout(100)
+ .setForwardWhenNoConsumers(forwardWhenNoConsumers)
+ .setMaxHops(maxHops)
+ .setConfirmationWindowSize(1024)
+ .setStaticConnectors(pairs);
conf.getClusterConfigurations().add(clusterConf);
}
@@ -2074,8 +2081,15 @@ public abstract class ClusterTestBase extends ServiceTestBase
final boolean forwardWhenNoConsumers, final int maxHops,
TransportConfiguration connectorFrom, List<String> pairs)
{
- return new ClusterConnectionConfiguration(name, address, connectorFrom.getName(), 250, true,
- forwardWhenNoConsumers, maxHops, 1024, pairs, false);
+ return new ClusterConnectionConfiguration()
+ .setName(name)
+ .setAddress(address)
+ .setConnectorName(connectorFrom.getName())
+ .setRetryInterval(250)
+ .setForwardWhenNoConsumers(forwardWhenNoConsumers)
+ .setMaxHops(maxHops)
+ .setConfirmationWindowSize(1024)
+ .setStaticConnectors(pairs);
}
protected void setupClusterConnectionWithBackups(final String name,
@@ -2104,9 +2118,16 @@ public abstract class ClusterTestBase extends ServiceTestBase
pairs.add(serverTotc.getName());
}
Configuration config = serverFrom.getConfiguration();
- ClusterConnectionConfiguration clusterConf =
- new ClusterConnectionConfiguration(name, address, name, 250, true, forwardWhenNoConsumers, maxHops,
- 1024, pairs, false);
+
+ ClusterConnectionConfiguration clusterConf = new ClusterConnectionConfiguration()
+ .setName(name)
+ .setAddress(address)
+ .setConnectorName(name)
+ .setRetryInterval(250)
+ .setForwardWhenNoConsumers(forwardWhenNoConsumers)
+ .setMaxHops(maxHops)
+ .setConfirmationWindowSize(1024)
+ .setStaticConnectors(pairs);
config.getClusterConfigurations().add(clusterConf);
}
@@ -2129,13 +2150,16 @@ public abstract class ClusterTestBase extends ServiceTestBase
TransportConfiguration connectorConfig = createTransportConfiguration(netty, false, generateParams(node, netty));
server.getConfiguration().getConnectorConfigurations().put(name, connectorConfig);
Configuration conf = server.getConfiguration();
- ClusterConnectionConfiguration clusterConf =
- new ClusterConnectionConfiguration(name, address, name, 100,
- true,
- forwardWhenNoConsumers,
- maxHops,
- 1024,
- discoveryGroupName);
+ ClusterConnectionConfiguration clusterConf = new ClusterConnectionConfiguration()
+ .setName(name)
+ .setAddress(address)
+ .setConnectorName(name)
+ .setRetryInterval(100)
+ .setDuplicateDetection(true)
+ .setForwardWhenNoConsumers(forwardWhenNoConsumers)
+ .setMaxHops(maxHops)
+ .setConfirmationWindowSize(1024)
+ .setDiscoveryGroupName(discoveryGroupName);
List<ClusterConnectionConfiguration> clusterConfs = conf.getClusterConfigurations();
clusterConfs.add(clusterConf);
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusterWithBackupTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusterWithBackupTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusterWithBackupTest.java
index 73a63db..f6099c4 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusterWithBackupTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusterWithBackupTest.java
@@ -117,9 +117,9 @@ public class ClusterWithBackupTest extends ClusterTestBase
setupBackupServer(2, 5, isFileStorage(), true, isNetty());
// The lives
- setupLiveServer(3, isFileStorage(), true, isNetty());
- setupLiveServer(4, isFileStorage(), true, isNetty());
- setupLiveServer(5, isFileStorage(), true, isNetty());
+ setupLiveServer(3, isFileStorage(), true, isNetty(), false);
+ setupLiveServer(4, isFileStorage(), true, isNetty(), false);
+ setupLiveServer(5, isFileStorage(), true, isNetty(), false);
}
}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java
index 64bc03b..895595e 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java
@@ -29,7 +29,7 @@ import org.hornetq.api.core.client.ClientMessage;
import org.hornetq.api.core.client.ClientProducer;
import org.hornetq.api.core.client.ClientSession;
import org.hornetq.api.core.client.ClientSessionFactory;
-import org.hornetq.api.core.management.NotificationType;
+import org.hornetq.api.core.management.CoreNotificationType;
import org.hornetq.core.postoffice.impl.BindingsImpl;
import org.hornetq.core.server.group.GroupingHandler;
import org.hornetq.core.server.group.UnproposalListener;
@@ -98,7 +98,8 @@ public class ClusteredGroupingTest extends ClusterTestBase
@Override
public void onNotification(Notification notification)
{
- if (notification.getType() == NotificationType.UNPROPOSAL)
+ if (!(notification.getType() instanceof CoreNotificationType)) return;
+ if (notification.getType() == CoreNotificationType.UNPROPOSAL)
{
latch.countDown();
}
@@ -109,7 +110,8 @@ public class ClusteredGroupingTest extends ClusterTestBase
@Override
public void onNotification(Notification notification)
{
- if (notification.getType() == NotificationType.UNPROPOSAL)
+ if (!(notification.getType() instanceof CoreNotificationType)) return;
+ if (notification.getType() == CoreNotificationType.UNPROPOSAL)
{
latch.countDown();
}
@@ -205,7 +207,8 @@ public class ClusteredGroupingTest extends ClusterTestBase
@Override
public void onNotification(Notification notification)
{
- if (notification.getType() == NotificationType.UNPROPOSAL)
+ if (!(notification.getType() instanceof CoreNotificationType)) return;
+ if (notification.getType() == CoreNotificationType.UNPROPOSAL)
{
latch.countDown();
}
@@ -216,7 +219,8 @@ public class ClusteredGroupingTest extends ClusterTestBase
@Override
public void onNotification(Notification notification)
{
- if (notification.getType() == NotificationType.UNPROPOSAL)
+ if (!(notification.getType() instanceof CoreNotificationType)) return;
+ if (notification.getType() == CoreNotificationType.UNPROPOSAL)
{
latch.countDown();
}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java
index f35b5b9..bff10b5 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java
@@ -76,10 +76,14 @@ public class MessageRedistributionTest extends ClusterTestBase
MessageRedistributionTest.log.info("Doing test");
- getServer(0).getConfiguration().setGroupingHandlerConfiguration(
- new GroupingHandlerConfiguration(new SimpleString("handler"), GroupingHandlerConfiguration.TYPE.LOCAL, new SimpleString("queues")));
- getServer(1).getConfiguration().setGroupingHandlerConfiguration(
- new GroupingHandlerConfiguration(new SimpleString("handler"), GroupingHandlerConfiguration.TYPE.REMOTE, new SimpleString("queues")));
+ getServer(0).getConfiguration().setGroupingHandlerConfiguration(new GroupingHandlerConfiguration()
+ .setName(new SimpleString("handler"))
+ .setType(GroupingHandlerConfiguration.TYPE.LOCAL)
+ .setAddress(new SimpleString("queues")));
+ getServer(1).getConfiguration().setGroupingHandlerConfiguration(new GroupingHandlerConfiguration()
+ .setName(new SimpleString("handler"))
+ .setType(GroupingHandlerConfiguration.TYPE.REMOTE)
+ .setAddress(new SimpleString("queues")));
startServers(0, 1);
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SimpleSymmetricClusterTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SimpleSymmetricClusterTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SimpleSymmetricClusterTest.java
index 6a263b3..7578fb3 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SimpleSymmetricClusterTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SimpleSymmetricClusterTest.java
@@ -67,9 +67,9 @@ public class SimpleSymmetricClusterTest extends ClusterTestBase
setupBackupServer(2, 5, isFileStorage(), true, isNetty());
// The lives
- setupLiveServer(3, isFileStorage(), true, isNetty());
- setupLiveServer(4, isFileStorage(), true, isNetty());
- setupLiveServer(5, isFileStorage(), true, isNetty());
+ setupLiveServer(3, isFileStorage(), true, isNetty(), false);
+ setupLiveServer(4, isFileStorage(), true, isNetty(), false);
+ setupLiveServer(5, isFileStorage(), true, isNetty(), false);
setupClusterConnection("cluster0", "queues", false, 1, isNetty(), 3, 4, 5);
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterTest.java
index e750d6c..58ed3e8 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterTest.java
@@ -25,6 +25,7 @@ import org.hornetq.tests.util.UnitTestCase;
* Most of the cases are covered in OneWayTwoNodeClusterTest - we don't duplicate them all here
*
* @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
+ * @author <a href="mtaylor@redhat,com">Martyn Taylor</a>
*
* Created 3 Feb 2009 09:10:43
*
@@ -1760,6 +1761,196 @@ public class SymmetricClusterTest extends ClusterTestBase
waitForBindings(4, "queues.testaddress", 0, 0, false);
}
+ @Test
+ /**
+ * This test verifies that addresses matching a simple string filter such as 'jms' result in bindings being created
+ * on appropriate nodes in the cluster. It also verifies that addresses not matching the simple string filter do not
+ * result in bindings being created.
+ */
+ public void testClusterAddressCreatesBindingsForSimpleStringAddressFilters() throws Exception
+ {
+ setupCluster("jms", "jms", "jms", "jms", "jms");
+ startServers();
+
+ setupSessionFactory(0, isNetty());
+ setupSessionFactory(1, isNetty());
+ setupSessionFactory(2, isNetty());
+ setupSessionFactory(3, isNetty());
+ setupSessionFactory(4, isNetty());
+
+ createQueue(0, "jms.queues.test.1", "queue0", null, false);
+ createQueue(1, "jms.queues.test.1", "queue0", null, false);
+ createQueue(2, "jms.queues.test.1", "queue0", null, false);
+ createQueue(3, "jms.queues.test.1", "queue0", null, false);
+ createQueue(4, "jms.queues.test.1", "queue0", null, false);
+
+ createQueue(0, "foo.queues.test.1", "queue1", null, false);
+ createQueue(1, "foo.queues.test.1", "queue1", null, false);
+ createQueue(2, "foo.queues.test.1", "queue1", null, false);
+ createQueue(3, "foo.queues.test.1", "queue1", null, false);
+ createQueue(4, "foo.queues.test.1", "queue1", null, false);
+
+ waitForBindings(0, "jms.queues.test.1", 4, 0, false);
+ waitForBindings(1, "jms.queues.test.1", 4, 0, false);
+ waitForBindings(2, "jms.queues.test.1", 4, 0, false);
+ waitForBindings(3, "jms.queues.test.1", 4, 0, false);
+ waitForBindings(4, "jms.queues.test.1", 4, 0, false);
+
+ waitForBindings(0, "foo.queues.test.1", 0, 0, false);
+ waitForBindings(1, "foo.queues.test.1", 0, 0, false);
+ waitForBindings(2, "foo.queues.test.1", 0, 0, false);
+ waitForBindings(3, "foo.queues.test.1", 0, 0, false);
+ waitForBindings(4, "foo.queues.test.1", 0, 0, false);
+ }
+
+ @Test
+ /**
+ * This test verifies that an string exclude filter '!jms.eu.uk' results in bindings not being created for this
+ * address for nodes in a cluster. But ensures that other addresses are matched and bindings created.
+ */
+ public void testClusterAddressDoesNotCreatesBindingsForStringExcludesAddressFilters() throws Exception
+ {
+ setupCluster("jms.eu.de,!jms.eu.uk", "jms.eu.de,!jms.eu.uk", "jms.eu.de,!jms.eu.uk", "jms.eu.de,!jms.eu.uk", "jms.eu.de,!jms.eu.uk");
+ startServers();
+
+ setupSessionFactory(0, isNetty());
+ setupSessionFactory(1, isNetty());
+ setupSessionFactory(2, isNetty());
+ setupSessionFactory(3, isNetty());
+ setupSessionFactory(4, isNetty());
+
+ createQueue(0, "jms.eu.uk", "queue0", null, false);
+ createQueue(1, "jms.eu.uk", "queue0", null, false);
+ createQueue(2, "jms.eu.uk", "queue0", null, false);
+ createQueue(3, "jms.eu.uk", "queue0", null, false);
+ createQueue(4, "jms.eu.uk", "queue0", null, false);
+
+ createQueue(0, "jms.eu.de", "queue1", null, false);
+ createQueue(1, "jms.eu.de", "queue1", null, false);
+ createQueue(2, "jms.eu.de", "queue1", null, false);
+ createQueue(3, "jms.eu.de", "queue1", null, false);
+ createQueue(4, "jms.eu.de", "queue1", null, false);
+
+ waitForBindings(0, "jms.eu.de", 4, 0, false);
+ waitForBindings(1, "jms.eu.de", 4, 0, false);
+ waitForBindings(2, "jms.eu.de", 4, 0, false);
+ waitForBindings(3, "jms.eu.de", 4, 0, false);
+ waitForBindings(4, "jms.eu.de", 4, 0, false);
+
+ waitForBindings(0, "jms.eu.uk", 0, 0, false);
+ waitForBindings(1, "jms.eu.uk", 0, 0, false);
+ waitForBindings(2, "jms.eu.uk", 0, 0, false);
+ waitForBindings(3, "jms.eu.uk", 0, 0, false);
+ waitForBindings(4, "jms.eu.uk", 0, 0, false);
+ }
+
+ /**
+ * This test verifies that remote bindings are only created for queues that match jms.eu or jms.us excluding
+ * jms.eu.uk and jms.us.bos. Represented by the address filter 'jms.eu,!jms.eu.uk,jms.us,!jms.us.bos'
+ * @throws Exception
+ */
+ @Test
+ public void testClusterAddressFiltersExcludesAndIncludesAddressesInList() throws Exception
+ {
+ setupCluster("jms.eu,!jms.eu.uk,jms.us,!jms.us.bos",
+ "jms.eu,!jms.eu.uk,jms.us,!jms.us.bos",
+ "jms.eu,!jms.eu.uk,jms.us,!jms.us.bos",
+ "jms.eu,!jms.eu.uk,jms.us,!jms.us.bos",
+ "jms.eu,!jms.eu.uk,jms.us,!jms.us.bos");
+
+ startServers();
+
+ setupSessionFactory(0, isNetty());
+ setupSessionFactory(1, isNetty());
+ setupSessionFactory(2, isNetty());
+ setupSessionFactory(3, isNetty());
+ setupSessionFactory(4, isNetty());
+
+ createQueue(0, "jms.eu.uk", "queue0", null, false);
+ createQueue(1, "jms.eu.uk", "queue0", null, false);
+ createQueue(2, "jms.eu.uk", "queue0", null, false);
+ createQueue(3, "jms.eu.uk", "queue0", null, false);
+ createQueue(4, "jms.eu.uk", "queue0", null, false);
+
+ createQueue(0, "jms.eu.de", "queue1", null, false);
+ createQueue(1, "jms.eu.de", "queue1", null, false);
+ createQueue(2, "jms.eu.de", "queue1", null, false);
+ createQueue(3, "jms.eu.de", "queue1", null, false);
+ createQueue(4, "jms.eu.de", "queue1", null, false);
+
+ createQueue(0, "jms.eu.fr", "queue2", null, false);
+ createQueue(1, "jms.eu.fr", "queue2", null, false);
+ createQueue(2, "jms.eu.fr", "queue2", null, false);
+ createQueue(3, "jms.eu.fr", "queue2", null, false);
+ createQueue(4, "jms.eu.fr", "queue2", null, false);
+
+ createQueue(0, "jms.us.ca", "queue4", null, false);
+ createQueue(1, "jms.us.ca", "queue4", null, false);
+ createQueue(2, "jms.us.ca", "queue4", null, false);
+ createQueue(3, "jms.us.ca", "queue4", null, false);
+ createQueue(4, "jms.us.ca", "queue4", null, false);
+
+ createQueue(0, "jms.us.se", "queue5", null, false);
+ createQueue(1, "jms.us.se", "queue5", null, false);
+ createQueue(2, "jms.us.se", "queue5", null, false);
+ createQueue(3, "jms.us.se", "queue5", null, false);
+ createQueue(4, "jms.us.se", "queue5", null, false);
+
+ createQueue(0, "jms.us.ny", "queue6", null, false);
+ createQueue(1, "jms.us.ny", "queue6", null, false);
+ createQueue(2, "jms.us.ny", "queue6", null, false);
+ createQueue(3, "jms.us.ny", "queue6", null, false);
+ createQueue(4, "jms.us.ny", "queue6", null, false);
+
+ waitForBindings(0, "jms.eu.de", 4, 0, false);
+ waitForBindings(1, "jms.eu.de", 4, 0, false);
+ waitForBindings(2, "jms.eu.de", 4, 0, false);
+ waitForBindings(3, "jms.eu.de", 4, 0, false);
+ waitForBindings(4, "jms.eu.de", 4, 0, false);
+
+ waitForBindings(0, "jms.eu.fr", 4, 0, false);
+ waitForBindings(1, "jms.eu.fr", 4, 0, false);
+ waitForBindings(2, "jms.eu.fr", 4, 0, false);
+ waitForBindings(3, "jms.eu.fr", 4, 0, false);
+ waitForBindings(4, "jms.eu.fr", 4, 0, false);
+
+ waitForBindings(0, "jms.eu.uk", 0, 0, false);
+ waitForBindings(1, "jms.eu.uk", 0, 0, false);
+ waitForBindings(2, "jms.eu.uk", 0, 0, false);
+ waitForBindings(3, "jms.eu.uk", 0, 0, false);
+ waitForBindings(4, "jms.eu.uk", 0, 0, false);
+
+ waitForBindings(0, "jms.us.ca", 4, 0, false);
+ waitForBindings(1, "jms.us.ca", 4, 0, false);
+ waitForBindings(2, "jms.us.ca", 4, 0, false);
+ waitForBindings(3, "jms.us.ca", 4, 0, false);
+ waitForBindings(4, "jms.us.ca", 4, 0, false);
+
+ waitForBindings(0, "jms.us.ny", 4, 0, false);
+ waitForBindings(1, "jms.us.ny", 4, 0, false);
+ waitForBindings(2, "jms.us.ny", 4, 0, false);
+ waitForBindings(3, "jms.us.ny", 4, 0, false);
+ waitForBindings(4, "jms.us.ny", 4, 0, false);
+
+ waitForBindings(0, "jms.us.bos", 0, 0, false);
+ waitForBindings(1, "jms.us.bos", 0, 0, false);
+ waitForBindings(2, "jms.us.bos", 0, 0, false);
+ waitForBindings(3, "jms.us.bos", 0, 0, false);
+ waitForBindings(4, "jms.us.bos", 0, 0, false);
+ }
+
+ protected void setupCluster(String addr1, String addr2, String addr3, String addr4, String addr5) throws Exception
+ {
+ setupClusterConnection("cluster0", addr1, true, 1, isNetty(), 0, 1, 2, 3, 4);
+
+ setupClusterConnection("cluster1", addr2, true, 1, isNetty(), 1, 0, 2, 3, 4);
+
+ setupClusterConnection("cluster2", addr3, true, 1, isNetty(), 2, 0, 1, 3, 4);
+
+ setupClusterConnection("cluster3", addr4, true, 1, isNetty(), 3, 0, 1, 2, 4);
+ setupClusterConnection("cluster4", addr5, true, 1, isNetty(), 4, 0, 1, 2, 3);
+ }
+
protected void setupCluster() throws Exception
{
setupCluster(false);
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterWithBackupTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterWithBackupTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterWithBackupTest.java
index b65c547..4dc5af7 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterWithBackupTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterWithBackupTest.java
@@ -12,7 +12,7 @@
*/
package org.hornetq.tests.integration.cluster.distribution;
-import org.hornetq.core.server.cluster.ha.HAPolicy;
+import org.hornetq.core.config.ha.SharedStoreSlavePolicyConfiguration;
import org.hornetq.tests.integration.IntegrationTestLogger;
import org.hornetq.tests.util.UnitTestCase;
import org.junit.Test;
@@ -539,11 +539,11 @@ public class SymmetricClusterWithBackupTest extends SymmetricClusterTest
setupBackupServer(9, 4, isFileStorage(), true, isNetty());
// The lives
- setupLiveServer(0, isFileStorage(), true, isNetty());
- setupLiveServer(1, isFileStorage(), true, isNetty());
- setupLiveServer(2, isFileStorage(), true, isNetty());
- setupLiveServer(3, isFileStorage(), true, isNetty());
- setupLiveServer(4, isFileStorage(), true, isNetty());
+ setupLiveServer(0, isFileStorage(), true, isNetty(), false);
+ setupLiveServer(1, isFileStorage(), true, isNetty(), false);
+ setupLiveServer(2, isFileStorage(), true, isNetty(), false);
+ setupLiveServer(3, isFileStorage(), true, isNetty(), false);
+ setupLiveServer(4, isFileStorage(), true, isNetty(), false);
}
@Override
@@ -551,11 +551,11 @@ public class SymmetricClusterWithBackupTest extends SymmetricClusterTest
{
// Need to set backup, since when restarting backup after it has failed over, backup will have been set to false
- getServer(5).getConfiguration().getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_SHARED_STORE);
- getServer(6).getConfiguration().getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_SHARED_STORE);
- getServer(7).getConfiguration().getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_SHARED_STORE);
- getServer(8).getConfiguration().getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_SHARED_STORE);
- getServer(9).getConfiguration().getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_SHARED_STORE);
+ getServer(5).getConfiguration().setHAPolicyConfiguration(new SharedStoreSlavePolicyConfiguration());
+ getServer(6).getConfiguration().setHAPolicyConfiguration(new SharedStoreSlavePolicyConfiguration());
+ getServer(7).getConfiguration().setHAPolicyConfiguration(new SharedStoreSlavePolicyConfiguration());
+ getServer(8).getConfiguration().setHAPolicyConfiguration(new SharedStoreSlavePolicyConfiguration());
+ getServer(9).getConfiguration().setHAPolicyConfiguration(new SharedStoreSlavePolicyConfiguration());
startServers(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/AutomaticColocatedQuorumVoteTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/AutomaticColocatedQuorumVoteTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/AutomaticColocatedQuorumVoteTest.java
index 0023973..52ca1fb 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/AutomaticColocatedQuorumVoteTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/AutomaticColocatedQuorumVoteTest.java
@@ -21,17 +21,22 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.hornetq.api.config.HornetQDefaultConfiguration;
import org.hornetq.api.core.TransportConfiguration;
import org.hornetq.api.core.client.ClientSessionFactory;
import org.hornetq.api.core.client.HornetQClient;
import org.hornetq.api.core.client.ServerLocator;
import org.hornetq.core.client.impl.Topology;
import org.hornetq.core.client.impl.TopologyMemberImpl;
-import org.hornetq.core.config.BackupStrategy;
import org.hornetq.core.config.Configuration;
import org.hornetq.core.config.CoreQueueConfiguration;
+import org.hornetq.core.config.ScaleDownConfiguration;
+import org.hornetq.core.config.ha.ColocatedPolicyConfiguration;
+import org.hornetq.core.config.ha.ReplicaPolicyConfiguration;
+import org.hornetq.core.config.ha.ReplicatedPolicyConfiguration;
+import org.hornetq.core.config.ha.SharedStoreMasterPolicyConfiguration;
+import org.hornetq.core.config.ha.SharedStoreSlavePolicyConfiguration;
import org.hornetq.core.server.HornetQServer;
-import org.hornetq.core.server.cluster.ha.HAPolicy;
import org.hornetq.core.server.impl.HornetQServerImpl;
import org.hornetq.tests.util.ServiceTestBase;
import org.junit.Test;
@@ -41,27 +46,27 @@ import org.junit.runners.Parameterized;
@RunWith(value = Parameterized.class)
public class AutomaticColocatedQuorumVoteTest extends ServiceTestBase
{
- private final HAPolicy.POLICY_TYPE policyType;
+ private final boolean replicated;
- @Parameterized.Parameters
+ @Parameterized.Parameters(name = "replicated={0}")
public static Collection getParameters()
{
return Arrays.asList(new Object[][]
{
- {HAPolicy.POLICY_TYPE.COLOCATED_REPLICATED},
- {HAPolicy.POLICY_TYPE.COLOCATED_SHARED_STORE}
+ {true},
+ {false}
});
}
- public AutomaticColocatedQuorumVoteTest(HAPolicy.POLICY_TYPE policyType)
+ public AutomaticColocatedQuorumVoteTest(boolean replicated)
{
- this.policyType = policyType;
+ this.replicated = replicated;
}
@Test
public void testSimpleDistributionBackupStrategyFull() throws Exception
{
- HornetQServer server0 = createServer(0, 1, BackupStrategy.FULL);
- HornetQServer server1 = createServer(1, 0, BackupStrategy.FULL);
+ HornetQServer server0 = createServer(0, 1, false);
+ HornetQServer server1 = createServer(1, 0, false);
TransportConfiguration liveConnector0 = getConnectorTransportConfiguration("liveConnector" + 0, 0);
TransportConfiguration liveConnector1 = getConnectorTransportConfiguration("liveConnector" + 1, 1);
@@ -103,7 +108,7 @@ public class AutomaticColocatedQuorumVoteTest extends ServiceTestBase
assertEquals(2, connectorConfigurations1.size());
assertEquals("5546", connectorConfigurations1.get("liveConnector1").getParams().get("port"));
assertEquals("5445", connectorConfigurations1.get("remoteConnector1").getParams().get("port"));
- if (policyType == HAPolicy.POLICY_TYPE.COLOCATED_SHARED_STORE)
+ if (!replicated)
{
assertEquals(server0.getConfiguration().getJournalDirectory(), backupServer1.getConfiguration().getJournalDirectory());
assertEquals(server0.getConfiguration().getBindingsDirectory(), backupServer1.getConfiguration().getBindingsDirectory());
@@ -128,7 +133,14 @@ public class AutomaticColocatedQuorumVoteTest extends ServiceTestBase
}
finally
{
- server0.stop();
+ try
+ {
+ server0.stop();
+ }
+ catch (Throwable e)
+ {
+ e.printStackTrace();
+ }
server1.stop();
}
}
@@ -136,8 +148,8 @@ public class AutomaticColocatedQuorumVoteTest extends ServiceTestBase
@Test
public void testSimpleDistributionBackupStrategyScaleDown() throws Exception
{
- HornetQServer server0 = createServer(0, 1, BackupStrategy.SCALE_DOWN);
- HornetQServer server1 = createServer(1, 0, BackupStrategy.SCALE_DOWN);
+ HornetQServer server0 = createServer(0, 1, true);
+ HornetQServer server1 = createServer(1, 0, true);
TransportConfiguration liveConnector0 = getConnectorTransportConfiguration("liveConnector" + 0, 0);
TransportConfiguration liveConnector1 = getConnectorTransportConfiguration("liveConnector" + 1, 1);
@@ -177,7 +189,7 @@ public class AutomaticColocatedQuorumVoteTest extends ServiceTestBase
assertEquals(2, connectorConfigurations1.size());
assertEquals("5446", connectorConfigurations1.get("liveConnector1").getParams().get("port"));
assertEquals("5445", connectorConfigurations1.get("remoteConnector1").getParams().get("port"));
- if (policyType == HAPolicy.POLICY_TYPE.COLOCATED_SHARED_STORE)
+ if (!replicated)
{
assertEquals(server0.getConfiguration().getJournalDirectory(), backupServer1.getConfiguration().getJournalDirectory());
assertEquals(server0.getConfiguration().getBindingsDirectory(), backupServer1.getConfiguration().getBindingsDirectory());
@@ -202,7 +214,14 @@ public class AutomaticColocatedQuorumVoteTest extends ServiceTestBase
}
finally
{
- server0.stop();
+ try
+ {
+ server0.stop();
+ }
+ catch (Throwable e)
+ {
+ e.printStackTrace();
+ }
server1.stop();
}
}
@@ -210,10 +229,10 @@ public class AutomaticColocatedQuorumVoteTest extends ServiceTestBase
@Test
public void testSimpleDistributionOfBackupsMaxBackupsExceeded() throws Exception
{
- HornetQServer server0 = createServer(0, 1, BackupStrategy.FULL);
- HornetQServer server1 = createServer(1, 0, BackupStrategy.FULL);
- HornetQServer server2 = createServer(2, 0, BackupStrategy.FULL);
- HornetQServer server3 = createServer(3, 0, BackupStrategy.FULL);
+ HornetQServer server0 = createServer(0, 1, false);
+ HornetQServer server1 = createServer(1, 0, false);
+ HornetQServer server2 = createServer(2, 0, false);
+ HornetQServer server3 = createServer(3, 0, false);
TransportConfiguration liveConnector0 = getConnectorTransportConfiguration("liveConnector" + 0, 0);
TransportConfiguration liveConnector1 = getConnectorTransportConfiguration("liveConnector" + 1, 1);
TransportConfiguration liveConnector2 = getConnectorTransportConfiguration("liveConnector" + 2, 2);
@@ -274,45 +293,69 @@ public class AutomaticColocatedQuorumVoteTest extends ServiceTestBase
}
}
- private HornetQServer createServer(int node, int remoteNode, BackupStrategy backupStrategy) throws Exception
+ private HornetQServer createServer(int node, int remoteNode, boolean scaleDown) throws Exception
{
TransportConfiguration liveConnector = getConnectorTransportConfiguration("liveConnector" + node, node);
TransportConfiguration remoteConnector = getConnectorTransportConfiguration("remoteConnector" + node, remoteNode);
TransportConfiguration liveAcceptor = getAcceptorTransportConfiguration(node);
- Configuration liveConfiguration = getConfiguration("server" + node, backupStrategy, liveConnector, liveAcceptor, remoteConnector);
+ Configuration liveConfiguration = getConfiguration("server" + node, scaleDown, liveConnector, liveAcceptor, remoteConnector);
HornetQServer server = new HornetQServerImpl(liveConfiguration);
server.setIdentity("server" + node);
return server;
}
- private Configuration getConfiguration(String identity, BackupStrategy backupStrategy, TransportConfiguration liveConnector, TransportConfiguration liveAcceptor, TransportConfiguration... otherLiveNodes) throws Exception
+
+ private Configuration getConfiguration(String identity, boolean scaleDown, TransportConfiguration liveConnector, TransportConfiguration liveAcceptor, TransportConfiguration... otherLiveNodes) throws Exception
{
- Configuration configuration = createDefaultConfig();
- configuration.getAcceptorConfigurations().clear();
- configuration.getAcceptorConfigurations().add(liveAcceptor);
- configuration.getConnectorConfigurations().put(liveConnector.getName(), liveConnector);
- configuration.setJournalDirectory(configuration.getJournalDirectory() + identity);
- configuration.setBindingsDirectory(configuration.getBindingsDirectory() + identity);
- configuration.setLargeMessagesDirectory(configuration.getLargeMessagesDirectory() + identity);
- configuration.setPagingDirectory(configuration.getPagingDirectory() + identity);
+ Configuration configuration = createDefaultConfig()
+ .clearAcceptorConfigurations()
+ .addAcceptorConfiguration(liveAcceptor)
+ .addConnectorConfiguration(liveConnector.getName(), liveConnector)
+ .setJournalDirectory(HornetQDefaultConfiguration.getDefaultJournalDir() + identity)
+ .setBindingsDirectory(HornetQDefaultConfiguration.getDefaultBindingsDirectory() + identity)
+ .setLargeMessagesDirectory(HornetQDefaultConfiguration.getDefaultLargeMessagesDir() + identity)
+ .setPagingDirectory(HornetQDefaultConfiguration.getDefaultPagingDir() + identity)
+ .addQueueConfiguration(new CoreQueueConfiguration()
+ .setAddress("jms.queue.testQueue")
+ .setName("jms.queue.testQueue"));
+
List<String> transportConfigurationList = new ArrayList<>();
- final HAPolicy haPolicy = new HAPolicy();
+
+ final ColocatedPolicyConfiguration haPolicy = new ColocatedPolicyConfiguration();
for (TransportConfiguration otherLiveNode : otherLiveNodes)
{
- configuration.getConnectorConfigurations().put(otherLiveNode.getName(), otherLiveNode);
+ configuration.addConnectorConfiguration(otherLiveNode.getName(), otherLiveNode);
transportConfigurationList.add(otherLiveNode.getName());
- haPolicy.getRemoteConnectors().add(otherLiveNode.getName());
+ haPolicy.getExcludedConnectors().add(otherLiveNode.getName());
}
- basicClusterConnectionConfig(configuration, liveConnector.getName(), transportConfigurationList);
- configuration.getQueueConfigurations().add(new CoreQueueConfiguration("jms.queue.testQueue", "jms.queue.testQueue", null, true));
-
- haPolicy.setPolicyType(policyType);
- haPolicy.setBackupStrategy(backupStrategy);
+ configuration.addClusterConfiguration(basicClusterConnectionConfig(liveConnector.getName(), transportConfigurationList));
haPolicy.setBackupPortOffset(100);
haPolicy.setBackupRequestRetries(-1);
haPolicy.setBackupRequestRetryInterval(500);
haPolicy.setMaxBackups(1);
haPolicy.setRequestBackup(true);
- configuration.setHAPolicy(haPolicy);
+ configuration.setHAPolicyConfiguration(haPolicy);
+ if (!replicated)
+ {
+ SharedStoreMasterPolicyConfiguration ssmc = new SharedStoreMasterPolicyConfiguration();
+ SharedStoreSlavePolicyConfiguration sssc = new SharedStoreSlavePolicyConfiguration();
+ haPolicy.setLiveConfig(ssmc);
+ haPolicy.setBackupConfig(sssc);
+ if (scaleDown)
+ {
+ sssc.setScaleDownConfiguration(new ScaleDownConfiguration());
+ }
+ }
+ else
+ {
+ ReplicatedPolicyConfiguration rpc = new ReplicatedPolicyConfiguration();
+ ReplicaPolicyConfiguration rpc2 = new ReplicaPolicyConfiguration();
+ haPolicy.setLiveConfig(rpc);
+ haPolicy.setBackupConfig(rpc2);
+ if (scaleDown)
+ {
+ rpc2.setScaleDownConfiguration(new ScaleDownConfiguration());
+ }
+ }
return configuration;
}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java
index daeafeb..891be38 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java
@@ -41,6 +41,7 @@ import org.hornetq.tests.integration.cluster.util.BackupSyncDelay;
import org.hornetq.tests.integration.cluster.util.TestableServer;
import org.hornetq.tests.util.TransportConfigurationUtils;
import org.hornetq.utils.UUID;
+import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -81,6 +82,27 @@ public class BackupSyncJournalTest extends FailoverTestBase
}
+ @Override
+ @After
+ public void tearDown() throws Exception
+ {
+ try
+ {
+ File dir = new File(backupServer.getServer()
+ .getConfiguration()
+ .getLargeMessagesDirectory());
+ deleteDirectory(dir);
+ dir = new File(liveServer.getServer()
+ .getConfiguration()
+ .getLargeMessagesDirectory());
+ deleteDirectory(dir);
+ }
+ finally
+ {
+ super.tearDown();
+ }
+ }
+
@Test
public void testNodeID() throws Exception
{
@@ -246,8 +268,8 @@ public class BackupSyncJournalTest extends FailoverTestBase
}
/**
- * @throws FileNotFoundException
- * @throws IOException
+ * @throws java.io.FileNotFoundException
+ * @throws java.io.IOException
* @throws InterruptedException
*/
private void assertNodeIdWasSaved() throws Exception
@@ -309,14 +331,14 @@ public class BackupSyncJournalTest extends FailoverTestBase
assertNoMoreMessages();
sendMessages(session, producer, 2 * n_msgs);
- assertFalse("must NOT be a backup", liveServer.getServer().getConfiguration().getHAPolicy().isBackup());
- adaptLiveConfigForReplicatedFailBack(liveServer.getServer().getConfiguration());
+ assertFalse("must NOT be a backup", liveServer.getServer().getHAPolicy().isBackup());
+ adaptLiveConfigForReplicatedFailBack(liveServer);
liveServer.start();
waitForServer(liveServer.getServer());
- assertTrue("must have become a backup", liveServer.getServer().getConfiguration().getHAPolicy().isBackup());
+ assertTrue("must have become a backup", liveServer.getServer().getHAPolicy().isBackup());
assertTrue("Fail-back must initialize live!", liveServer.getServer().waitForActivation(15, TimeUnit.SECONDS));
- assertFalse("must be LIVE!", liveServer.getServer().getConfiguration().getHAPolicy().isBackup());
+ assertFalse("must be LIVE!", liveServer.getServer().getHAPolicy().isBackup());
int i = 0;
while (backupServer.isStarted() && i++ < 100)
{
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java
index 64469ac..ecde2ec 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java
@@ -62,6 +62,8 @@ public class BackupSyncLargeMessageTest extends BackupSyncJournalTest
@Test
public void testDeleteLargeMessages() throws Exception
{
+ // 200 will increase the odds of a failure
+ setNumberOfMessages(200);
File dir = new File(backupServer.getServer().getConfiguration().getLargeMessagesDirectory());
assertEquals("Should not have any large messages... previous test failed to clean up?", 0,
getAllMessageFileIds(dir).size());
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ColocatedFailoverCheckPairingsNettyTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ColocatedFailoverCheckPairingsNettyTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ColocatedFailoverCheckPairingsNettyTest.java
deleted file mode 100644
index 763fa48..0000000
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ColocatedFailoverCheckPairingsNettyTest.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * Copyright 2005-2014 Red Hat, Inc.
- * Red Hat licenses this file to you under the Apache License, version
- * 2.0 (the "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * http://www.apache.org/licenses/LICENSE-2.0
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
- * implied. See the License for the specific language governing
- * permissions and limitations under the License.
- */
-package org.hornetq.tests.integration.cluster.failover;
-
-
-import org.hornetq.api.core.TransportConfiguration;
-import org.hornetq.core.client.impl.TopologyMemberImpl;
-import org.hornetq.core.config.Configuration;
-import org.hornetq.core.config.CoreQueueConfiguration;
-import org.hornetq.core.server.cluster.ha.HAPolicy;
-import org.hornetq.core.server.impl.InVMNodeManager;
-
-import java.util.HashMap;
-
-public class ColocatedFailoverCheckPairingsNettyTest extends ColocatedFailoverCheckPairingsTest
-{
-
- protected void assertPairedCorrectly(TopologyMemberImpl m1, TopologyMemberImpl m2)
- {
- String backup1Port = (String) m1.getBackup().getParams().get("port");
- String backup2Port = (String) m2.getBackup().getParams().get("port");
- assertEquals(backup1Port, "5449");
- assertEquals(backup2Port, "5448");
- }
- @Override
- protected void createConfigs() throws Exception
- {
-
- TransportConfiguration liveConnector1 = getConnectorTransportConfiguration(1);
- liveConfiguration1 = super.createDefaultConfig();
- liveConfiguration1.getAcceptorConfigurations().clear();
- liveConfiguration1.getAcceptorConfigurations().add(getAcceptorTransportConfiguration(1));
- liveConfiguration1.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.REPLICATED);
- liveConfiguration1.getHAPolicy().setFailbackDelay(1000);
- liveConfiguration1.setJournalDirectory(getTestDir() + "/live1/journal");
- liveConfiguration1.setBindingsDirectory(getTestDir() + "/live1/bindings");
- liveConfiguration1.getQueueConfigurations().add(new CoreQueueConfiguration("jms.queue.testQueue", "jms.queue.testQueue", null, true));
-
- TransportConfiguration liveConnector2 = getConnectorTransportConfiguration(2);
- basicClusterConnectionConfig(liveConfiguration1, liveConnector1.getName(), liveConnector2.getName());
- liveConfiguration1.getConnectorConfigurations().put(liveConnector1.getName(), liveConnector1);
- liveConfiguration1.getConnectorConfigurations().put(liveConnector2.getName(), liveConnector2);
-
- Configuration backupConfiguration1 = liveConfiguration1.copy();
- TransportConfiguration backupConnector1 = getConnectorTransportConfiguration(3);
- backupConfiguration1.getConnectorConfigurations().put(backupConnector1.getName(), backupConnector1);
- backupConfiguration1.setJournalDirectory(getTestDir() + "/backup1/journal");
-
- backupConfiguration1.setBindingsDirectory(getTestDir() + "/backup1/bindings");
- backupConfiguration1.getAcceptorConfigurations().clear();
- backupConfiguration1.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_REPLICATED);
- backupConfiguration1.getClusterConfigurations().clear();
- basicClusterConnectionConfig(backupConfiguration1, backupConnector1.getName(), liveConnector1.getName());
- liveConfiguration1.getBackupServerConfigurations().add(backupConfiguration1);
-
-
- liveConfiguration2 = super.createDefaultConfig();
- liveConfiguration2.getAcceptorConfigurations().clear();
- liveConfiguration2.getAcceptorConfigurations().add(getAcceptorTransportConfiguration(2));
- liveConfiguration2.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.REPLICATED);
- liveConfiguration2.getHAPolicy().setFailbackDelay(1000);
- liveConfiguration2.setJournalDirectory(getTestDir() + "/live2/journal");
- liveConfiguration2.setBindingsDirectory(getTestDir() + "/live2/bindings");
- liveConfiguration2.getQueueConfigurations().add(new CoreQueueConfiguration("jms.queue.testQueue", "jms.queue.testQueue", null, true));
-
- basicClusterConnectionConfig(liveConfiguration2, liveConnector2.getName(), liveConnector1.getName());
- liveConfiguration2.getConnectorConfigurations().put(liveConnector1.getName(), liveConnector1);
- liveConfiguration2.getConnectorConfigurations().put(liveConnector2.getName(), liveConnector2);
-
- Configuration backupConfiguration2 = liveConfiguration2.copy();
- TransportConfiguration backupConnector2 = getConnectorTransportConfiguration(4);
- backupConfiguration2.getAcceptorConfigurations().clear();
- backupConfiguration2.getConnectorConfigurations().put(backupConnector2.getName(), backupConnector2);
- backupConfiguration2.setJournalDirectory(getTestDir() + "/backup2/journal");
- backupConfiguration2.setBindingsDirectory(getTestDir() + "/backup2/bindings");
- backupConfiguration2.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_REPLICATED);
- backupConfiguration2.getClusterConfigurations().clear();
- basicClusterConnectionConfig(backupConfiguration2, backupConnector2.getName(), liveConnector2.getName());
- liveConfiguration2.getBackupServerConfigurations().add(backupConfiguration2);
-
- nodeManagerLive1 = new InVMNodeManager(true, backupConfiguration2.getJournalDirectory());
- nodeManagerLive2 = new InVMNodeManager(true, backupConfiguration1.getJournalDirectory());
-
- liveServer1 = createTestableServer(liveConfiguration1, nodeManagerLive1, nodeManagerLive2, 1);
- liveServer2 = createTestableServer(liveConfiguration2, nodeManagerLive2, nodeManagerLive1, 2);
- }
-
- @Override
- protected TransportConfiguration getAcceptorTransportConfiguration(int node)
- {
- HashMap<String, Object> params = new HashMap<>();
- params.put("port", "" + (5445 + node));
- return new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, params);
- }
-
- @Override
- protected TransportConfiguration getConnectorTransportConfiguration(int node)
- {
- HashMap<String, Object> params = new HashMap<>();
- params.put("port", "" + (5445 + node));
- return new TransportConfiguration(NETTY_CONNECTOR_FACTORY, params);
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ColocatedFailoverCheckPairingsTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ColocatedFailoverCheckPairingsTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ColocatedFailoverCheckPairingsTest.java
deleted file mode 100644
index 976dbfb..0000000
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ColocatedFailoverCheckPairingsTest.java
+++ /dev/null
@@ -1,160 +0,0 @@
-/*
- * Copyright 2005-2014 Red Hat, Inc.
- * Red Hat licenses this file to you under the Apache License, version
- * 2.0 (the "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * http://www.apache.org/licenses/LICENSE-2.0
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
- * implied. See the License for the specific language governing
- * permissions and limitations under the License.
- */
-package org.hornetq.tests.integration.cluster.failover;
-
-
-import java.util.HashMap;
-
-import org.hornetq.api.core.TransportConfiguration;
-import org.hornetq.api.core.client.ClientSessionFactory;
-import org.hornetq.api.core.client.HornetQClient;
-import org.hornetq.api.core.client.ServerLocator;
-import org.hornetq.core.client.impl.Topology;
-import org.hornetq.core.client.impl.TopologyMemberImpl;
-import org.hornetq.core.config.Configuration;
-import org.hornetq.core.config.CoreQueueConfiguration;
-import org.hornetq.core.server.NodeManager;
-import org.hornetq.core.server.cluster.ha.HAPolicy;
-import org.hornetq.core.server.impl.InVMNodeManager;
-import org.hornetq.tests.integration.cluster.util.SameProcessHornetQServer;
-import org.hornetq.tests.integration.cluster.util.TestableServer;
-import org.hornetq.tests.util.ServiceTestBase;
-import org.junit.After;
-import org.junit.Test;
-
-public class ColocatedFailoverCheckPairingsTest extends ServiceTestBase
-{
- TestableServer liveServer1;
-
- protected TestableServer liveServer2;
-
- protected Configuration liveConfiguration1;
-
- protected Configuration liveConfiguration2;
-
- protected NodeManager nodeManagerLive1;
-
- protected NodeManager nodeManagerLive2;
-
- @Override
- public void setUp() throws Exception
- {
- super.setUp();
- }
-
- @Test
- public void testPairings() throws Exception
- {
- createConfigs();
-
- liveServer1.start();
- liveServer2.start();
- waitForServer(liveServer1.getServer());
- waitForServer(liveServer2.getServer());
-
- ServerLocator locator = HornetQClient.createServerLocator(true, getConnectorTransportConfiguration(1));
-
- waitForTopology(liveServer1.getServer(), 2, 2);
- waitForTopology(liveServer2.getServer(), 2, 2);
-
- ClientSessionFactory sessionFactory = locator.createSessionFactory();
-
- Topology topology = sessionFactory.getServerLocator().getTopology();
- System.out.println(topology.describe());
- TopologyMemberImpl m1 = topology.getMember(liveServer1.getServer().getNodeID().toString());
- TopologyMemberImpl m2 = topology.getMember(liveServer2.getServer().getNodeID().toString());
-
- assertPairedCorrectly(m1, m2);
- }
-
- protected void assertPairedCorrectly(TopologyMemberImpl m1, TopologyMemberImpl m2)
- {
- assertEquals(m1.getLive(), m2.getBackup());
- assertEquals(m2.getLive(), m1.getBackup());
- }
-
- @Override
- @After
- public void tearDown() throws Exception
- {
- liveServer1.stop();
- liveServer2.stop();
- super.tearDown();
- }
-
- protected void createConfigs() throws Exception
- {
- nodeManagerLive1 = new InVMNodeManager(false);
- nodeManagerLive2 = new InVMNodeManager(false);
-
- TransportConfiguration liveConnector1 = getConnectorTransportConfiguration(1);
- liveConfiguration1 = super.createDefaultConfig();
- liveConfiguration1.getAcceptorConfigurations().clear();
- liveConfiguration1.getAcceptorConfigurations().add(getAcceptorTransportConfiguration(1));
- liveConfiguration1.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.SHARED_STORE);
- liveConfiguration1.getHAPolicy().setFailbackDelay(1000);
- liveConfiguration1.setJournalDirectory(getTestDir() + "/live1");
- liveConfiguration1.getQueueConfigurations().add(new CoreQueueConfiguration("jms.queue.testQueue", "jms.queue.testQueue", null, true));
-
- TransportConfiguration liveConnector2 = getConnectorTransportConfiguration(2);
- basicClusterConnectionConfig(liveConfiguration1, liveConnector1.getName(), liveConnector2.getName());
- liveConfiguration1.getConnectorConfigurations().put(liveConnector1.getName(), liveConnector1);
- liveConfiguration1.getConnectorConfigurations().put(liveConnector2.getName(), liveConnector2);
-
- Configuration backupConfiguration1 = liveConfiguration1.copy();
- backupConfiguration1.setJournalDirectory(getTestDir() + "/live2");
- backupConfiguration1.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_SHARED_STORE);
- liveConfiguration1.getBackupServerConfigurations().add(backupConfiguration1);
-
- liveServer1 = createTestableServer(liveConfiguration1, nodeManagerLive1, nodeManagerLive2, 1);
-
- liveConfiguration2 = super.createDefaultConfig();
- liveConfiguration2.getAcceptorConfigurations().clear();
- liveConfiguration2.getAcceptorConfigurations().add(getAcceptorTransportConfiguration(2));
- liveConfiguration2.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.SHARED_STORE);
- liveConfiguration2.getHAPolicy().setFailbackDelay(1000);
- liveConfiguration2.setJournalDirectory(getTestDir() + "/live2");
- liveConfiguration2.getQueueConfigurations().add(new CoreQueueConfiguration("jms.queue.testQueue", "jms.queue.testQueue", null, true));
-
- basicClusterConnectionConfig(liveConfiguration2, liveConnector2.getName(), liveConnector1.getName());
- liveConfiguration2.getConnectorConfigurations().put(liveConnector1.getName(), liveConnector1);
- liveConfiguration2.getConnectorConfigurations().put(liveConnector2.getName(), liveConnector2);
-
- Configuration backupConfiguration2 = liveConfiguration2.copy();
- backupConfiguration2.setJournalDirectory(getTestDir() + "/live1");
- backupConfiguration2.getHAPolicy().setPolicyType(HAPolicy.POLICY_TYPE.BACKUP_SHARED_STORE);
- liveConfiguration2.getBackupServerConfigurations().add(backupConfiguration2);
-
- liveServer2 = createTestableServer(liveConfiguration2, nodeManagerLive2, nodeManagerLive1, 2);
- }
-
- protected TransportConfiguration getAcceptorTransportConfiguration(int node)
- {
- HashMap<String, Object> params = new HashMap<>();
- params.put("server-id", "" + node);
- return new TransportConfiguration(INVM_ACCEPTOR_FACTORY, params);
- }
-
- protected TransportConfiguration getConnectorTransportConfiguration(int node)
- {
- HashMap<String, Object> params = new HashMap<>();
- params.put("server-id", "" + node);
- return new TransportConfiguration(INVM_CONNECTOR_FACTORY, params);
- }
-
- protected TestableServer createTestableServer(Configuration config, NodeManager liveNodeManager, NodeManager backupNodeManager, int id)
- {
- return new SameProcessHornetQServer(
- createColocatedInVMFailoverServer(true, config, liveNodeManager, backupNodeManager, id));
- }
-}