activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From martyntay...@apache.org
Subject activemq-artemis git commit: Added cluster tests for new route type
Date Mon, 24 Oct 2016 15:56:51 GMT
Repository: activemq-artemis
Updated Branches:
  refs/heads/ARTEMIS-780 6d41d37e9 -> bce5d4888


Added cluster tests for new route type


Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/bce5d488
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/bce5d488
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/bce5d488

Branch: refs/heads/ARTEMIS-780
Commit: bce5d48889dcb9fdfa19f7770d3cf9c11e92ae80
Parents: 6d41d37
Author: Martyn Taylor <mtaylor@redhat.com>
Authored: Mon Oct 24 16:56:30 2016 +0100
Committer: Martyn Taylor <mtaylor@redhat.com>
Committed: Mon Oct 24 16:56:30 2016 +0100

----------------------------------------------------------------------
 .../integration/addressing/AddressingTest.java  |  13 +-
 .../AnycastRoutingWithClusterTest.java          | 276 +++++++++++++++++++
 .../cluster/distribution/ClusterTestBase.java   |  14 +
 3 files changed, 297 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bce5d488/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
index 2e0fda4..03739e9 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
@@ -35,6 +35,7 @@ import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.impl.AddressInfo;
 import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 public class AddressingTest extends ActiveMQTestBase {
@@ -91,8 +92,6 @@ public class AddressingTest extends ActiveMQTestBase {
 
          q1.deleteQueue();
          q2.deleteQueue();
-
-         System.out.println(consumeAddress);
       }
    }
 
@@ -134,8 +133,6 @@ public class AddressingTest extends ActiveMQTestBase {
 
          q1.deleteQueue();
          q2.deleteQueue();
-
-         System.out.println(consumeAddress);
       }
    }
 
@@ -222,36 +219,40 @@ public class AddressingTest extends ActiveMQTestBase {
 
          q1.deleteQueue();
          q2.deleteQueue();
-
-         System.out.println(consumeAddress);
       }
    }
 
+   @Ignore
    @Test
    public void testDeleteQueueOnNoConsumersTrue() {
       fail("Not Implemented");
    }
 
+   @Ignore
    @Test
    public void testDeleteQueueOnNoConsumersFalse() {
       fail("Not Implemented");
    }
 
+   @Ignore
    @Test
    public void testLimitOnMaxConsumers() {
       fail("Not Implemented");
    }
 
+   @Ignore
    @Test
    public void testUnlimitedMaxConsumers() {
       fail("Not Implemented");
    }
 
+   @Ignore
    @Test
    public void testDefaultMaxConsumersFromAddress() {
       fail("Not Implemented");
    }
 
+   @Ignore
    @Test
    public void testDefaultDeleteOnNoConsumersFromAddress() {
       fail("Not Implemented");

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bce5d488/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/AnycastRoutingWithClusterTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/AnycastRoutingWithClusterTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/AnycastRoutingWithClusterTest.java
new file mode 100644
index 0000000..f413113
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/AnycastRoutingWithClusterTest.java
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activemq.artemis.tests.integration.cluster.distribution;
+
+import java.util.List;
+
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
+import org.apache.activemq.artemis.core.server.group.impl.GroupingHandlerConfiguration;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.junit.Test;
+
+public class AnycastRoutingWithClusterTest extends ClusterTestBase {
+
+   /**
+    * Test anycast address with single distributed queue in a 3 node cluster environment.
 Messages should be
+    * "round robin"'d across the each queue
+    * @throws Exception
+    */
+   @Test
+   public void testAnycastAddressOneQueueRoutingMultiNode() throws Exception {
+      String address = "test.address";
+      String queueName = "test.queue";
+      String clusterAddress = "test";
+
+      for (int i = 0; i < 3; i++) {
+         setupServer(i, isFileStorage(), isNetty());
+      }
+
+      setupClusterConnection("cluster0", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 0, 1, 2);
+      setupClusterConnection("cluster1", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 1, 0, 2);
+      setupClusterConnection("cluster2", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 2, 0, 1);
+
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.LOCAL, 0);
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 1);
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 2);
+
+      startServers(0, 1, 2);
+
+      List<Queue> queues;
+      for (int i = 0; i < 3; i++) {
+         createAddressInfo(i, address, AddressInfo.RoutingType.ANYCAST, -1, false);
+         setupSessionFactory(i, isNetty());
+         createQueue(i, address, queueName, null, false);
+         addConsumer(i, i, queueName, null);
+      }
+
+      for (int i = 0; i < 3; i++) {
+         waitForBindings(i, address, 1, 1, true);
+         waitForBindings(i, address, 2, 2, false);
+      }
+
+      final int noMessages = 30;
+      send(0, address, noMessages, true, null, null);
+
+      for (int s = 0; s < 3; s++) {
+         final Queue queue = servers[s].locateQueue(new SimpleString(queueName));
+         Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisfied() throws Exception {
+               return queue.getMessageCount() == noMessages / 3;
+            }
+         });
+      }
+
+      // Each consumer should receive noMessages / noServers
+      for (int i = 0; i < noMessages / 3; i++) {
+         for (int c = 0; c < 3; c++) {
+            assertNotNull(consumers[c].consumer.receive(1000));
+         }
+      }
+   }
+
+
+   /**
+    * Test anycast address with N queues in a 3 node cluster environment.  Messages should
be "round robin"'d across the
+    * each queue.
+    * @throws Exception
+    */
+   @Test
+   public void testAnycastAddressMultiQueuesRoutingMultiNode() throws Exception {
+
+      String address = "test.address";
+      String queueNamePrefix = "test.queue";
+      String clusterAddress = "test";
+
+      for (int i = 0; i < 3; i++) {
+         setupServer(i, isFileStorage(), isNetty());
+      }
+
+      setupClusterConnection("cluster0", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 0, 1, 2);
+      setupClusterConnection("cluster1", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 1, 0, 2);
+      setupClusterConnection("cluster2", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 2, 0, 1);
+
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.LOCAL, 0);
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 1);
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 2);
+
+      startServers(0, 1, 2);
+
+      List<Queue> queues;
+      for (int i = 0; i < 3; i++) {
+         createAddressInfo(i, address, AddressInfo.RoutingType.ANYCAST, -1, false);
+         setupSessionFactory(i, isNetty());
+         createQueue(i, address, queueNamePrefix + i, null, false);
+         addConsumer(i, i, queueNamePrefix + i, null);
+      }
+
+      for (int i = 0; i < 3; i++) {
+         waitForBindings(i, address, 1, 1, true);
+         waitForBindings(i, address, 2, 2, false);
+      }
+
+      final int noMessages = 30;
+      send(0, address, noMessages, true, null, null);
+
+      for (int s = 0; s < 3; s++) {
+         final Queue queue = servers[s].locateQueue(new SimpleString(queueNamePrefix + s));
+         Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisfied() throws Exception {
+               return queue.getMessageCount() == noMessages / 3;
+            }
+         });
+      }
+
+      // Each consumer should receive noMessages / noServers
+      for (int i = 0; i < noMessages / 3; i++) {
+         for (int c = 0; c < 3; c++) {
+            assertNotNull(consumers[c].consumer.receive(1000));
+         }
+      }
+   }
+
+   /**
+    * Test anycast address with N queues in a 3 node cluster environment.  Messages should
be "round robin"'d across the
+    * each queue.
+    * @throws Exception
+    */
+   @Test
+   public void testAnycastAddressMultiQueuesWithFilterRoutingMultiNode() throws Exception
{
+
+      String address = "test.address";
+      String queueNamePrefix = "test.queue";
+      String clusterAddress = "test";
+
+      for (int i = 0; i < 3; i++) {
+         setupServer(i, isFileStorage(), isNetty());
+      }
+
+      setupClusterConnection("cluster0", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 0, 1, 2);
+      setupClusterConnection("cluster1", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 1, 0, 2);
+      setupClusterConnection("cluster2", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 2, 0, 1);
+
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.LOCAL, 0);
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 1);
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 2);
+
+      startServers(0, 1, 2);
+
+      List<Queue> queues;
+      for (int i = 0; i < 3; i++) {
+         createAddressInfo(i, address, AddressInfo.RoutingType.ANYCAST, -1, false);
+         setupSessionFactory(i, isNetty());
+
+      }
+
+      String filter1 = "giraffe";
+      String filter2 = "platypus";
+
+      createQueue(0, address, queueNamePrefix + 0, filter1, false);
+      createQueue(1, address, queueNamePrefix + 1, filter1, false);
+      createQueue(2, address, queueNamePrefix + 2, filter2, false);
+
+      for (int i = 0; i < 3; i++) {
+         addConsumer(i, i, queueNamePrefix + i, null);
+      }
+
+      for (int i = 0; i < 3; i++) {
+         waitForBindings(i, address, 1, 1, true);
+         waitForBindings(i, address, 2, 2, false);
+      }
+
+      final int noMessages = 30;
+      send(0, address, noMessages, true, filter1, null);
+
+      // Each consumer should receive noMessages / noServers
+      for (int i = 0; i < noMessages / 2; i++) {
+         for (int c = 0; c < 2; c++) {
+            assertNotNull(consumers[c].consumer.receive(1000));
+         }
+      }
+
+      assertNull(consumers[2].consumer.receive(1000));
+   }
+
+   /**
+    * Test multicast address that with N queues in a 3 node cluster environment.  Each queue
should receive all messages
+    * sent from the client.
+    * @throws Exception
+    */
+   @Test
+   public void testMulitcastAddressMultiQueuesRoutingMultiNode() throws Exception {
+
+      String address = "test.address";
+      String queueNamePrefix = "test.queue";
+      String clusterAddress = "test";
+
+      for (int i = 0; i < 3; i++) {
+         setupServer(i, isFileStorage(), isNetty());
+      }
+
+      setupClusterConnection("cluster0", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 0, 1, 2);
+      setupClusterConnection("cluster1", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 1, 0, 2);
+      setupClusterConnection("cluster2", clusterAddress, MessageLoadBalancingType.STRICT,
1, isNetty(), 2, 0, 1);
+
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.LOCAL, 0);
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 1);
+      setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 2);
+
+      startServers(0, 1, 2);
+
+      List<Queue> queues;
+      for (int i = 0; i < 3; i++) {
+         createAddressInfo(i, address, AddressInfo.RoutingType.MULTICAST, -1, false);
+         setupSessionFactory(i, isNetty());
+         createQueue(i, address, queueNamePrefix + i, null, false);
+         addConsumer(i, i, queueNamePrefix + i, null);
+      }
+
+      for (int i = 0; i < 3; i++) {
+         waitForBindings(i, address, 1, 1, true);
+         waitForBindings(i, address, 2, 2, false);
+      }
+
+      final int noMessages = 30;
+      send(0, address, noMessages, true, null, null);
+
+      for (int s = 0; s < 3; s++) {
+         final Queue queue = servers[s].locateQueue(new SimpleString(queueNamePrefix + s));
+         Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisfied() throws Exception {
+               return queue.getMessageCount() == noMessages;
+            }
+         });
+      }
+
+      // Each consumer should receive noMessages
+      for (int i = 0; i < noMessages; i++) {
+         for (int c = 0; c < 3; c++) {
+            assertNotNull(consumers[c].consumer.receive(1000));
+         }
+      }
+   }
+
+   private boolean isNetty() {
+      return true;
+   }
+}

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bce5d488/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/ClusterTestBase.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/ClusterTestBase.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/ClusterTestBase.java
index 538779f..2623e9c 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/ClusterTestBase.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/ClusterTestBase.java
@@ -78,6 +78,7 @@ import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancing
 import org.apache.activemq.artemis.core.server.cluster.qourum.SharedNothingBackupQuorum;
 import org.apache.activemq.artemis.core.server.group.GroupingHandler;
 import org.apache.activemq.artemis.core.server.group.impl.GroupingHandlerConfiguration;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
 import org.apache.activemq.artemis.core.server.impl.InVMNodeManager;
 import org.apache.activemq.artemis.tests.integration.IntegrationTestLogger;
 import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
@@ -518,6 +519,19 @@ public abstract class ClusterTestBase extends ActiveMQTestBase {
       session.close();
    }
 
+   protected void createAddressInfo(final int node,
+                                    final String address,
+                                    final AddressInfo.RoutingType routingType,
+                                    final int defaulMaxConsumers,
+                                    boolean defaultDeleteOnNoConsumers) {
+      AddressInfo addressInfo = new AddressInfo(new SimpleString(address));
+      addressInfo.setRoutingType(routingType);
+      addressInfo.setDefaultMaxConsumers(defaulMaxConsumers);
+      addressInfo.setDefaultDeleteOnNoConsumers(defaultDeleteOnNoConsumers);
+
+      servers[node].createOrUpdateAddressInfo(addressInfo);
+   }
+
    protected void deleteQueue(final int node, final String queueName) throws Exception {
       ClientSessionFactory sf = sfs[node];
 


Mime
View raw message