You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2017/04/19 04:50:37 UTC
[3/3] activemq-artemis git commit: ARTEMIS-1123 Clean up and add new
AMQP tests
ARTEMIS-1123 Clean up and add new AMQP tests
Adds some new AMQP protocol handling tests brought forward from
ActiveMQ 5.x as well as cleaning up some of th existing tests
code to make adding some other tests easier.
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/0260a304
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/0260a304
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/0260a304
Branch: refs/heads/master
Commit: 0260a304b41ca582baacea7f10c03fde21405f7c
Parents: cc4c395
Author: Timothy Bish <ta...@gmail.com>
Authored: Tue Apr 18 16:50:08 2017 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Apr 19 00:50:13 2017 -0400
----------------------------------------------------------------------
.../transport/amqp/client/AmqpConnection.java | 20 +-
.../amqp/AmqpAnonymousRelayTest.java | 126 ++++++++
.../integration/amqp/AmqpClientTestSupport.java | 60 +++-
.../amqp/AmqpDeliveryAnnotationsTest.java | 8 +-
.../amqp/AmqpDescribedTypePayloadTest.java | 16 +-
.../amqp/AmqpExpiredMessageTest.java | 42 +--
.../amqp/AmqpFailoverEndpointDiscoveryTest.java | 135 +++++++++
.../amqp/AmqpInboundConnectionTest.java | 255 ++++++++++++++++
.../integration/amqp/AmqpManagementTest.java | 68 +++++
.../amqp/AmqpMessagePriorityTest.java | 36 +--
.../integration/amqp/AmqpNettyFailoverTest.java | 141 ---------
.../amqp/AmqpPresettledReceiverTest.java | 38 +--
.../amqp/AmqpReceiverDispositionTest.java | 19 +-
.../integration/amqp/AmqpReceiverDrainTest.java | 20 +-
.../amqp/AmqpReceiverWithFiltersTest.java | 12 +-
.../amqp/AmqpScheduledMessageTest.java | 193 +++++++++----
.../integration/amqp/AmqpSecurityTest.java | 14 +-
.../integration/amqp/AmqpSendReceiveTest.java | 114 ++++----
.../tests/integration/amqp/AmqpTestSupport.java | 12 +-
.../integration/amqp/AmqpTransactionTest.java | 90 +++---
.../tests/integration/amqp/ProtonTest.java | 289 +------------------
21 files changed, 1001 insertions(+), 707 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java
----------------------------------------------------------------------
diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java
index 76717fd..a30bae1 100644
--- a/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java
+++ b/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java
@@ -87,6 +87,7 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
private final String password;
private final URI remoteURI;
private final String connectionId;
+ private List<Symbol> desiredCapabilities = Collections.emptyList();
private List<Symbol> offeredCapabilities = Collections.emptyList();
private Map<Symbol, Object> offeredProperties = Collections.emptyMap();
@@ -146,6 +147,9 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
getEndpoint().setContainer(safeGetContainerId());
}
getEndpoint().setHostname(remoteURI.getHost());
+ if (!getDesiredCapabilities().isEmpty()) {
+ getEndpoint().setDesiredCapabilities(getDesiredCapabilities().toArray(new Symbol[0]));
+ }
if (!getOfferedCapabilities().isEmpty()) {
getEndpoint().setOfferedCapabilities(getOfferedCapabilities().toArray(new Symbol[0]));
}
@@ -393,12 +397,24 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
this.drainTimeout = drainTimeout;
}
+ public List<Symbol> getDesiredCapabilities() {
+ return desiredCapabilities;
+ }
+
+ public void setDesiredCapabilities(List<Symbol> desiredCapabilities) {
+ if (desiredCapabilities == null) {
+ desiredCapabilities = Collections.emptyList();
+ }
+
+ this.desiredCapabilities = desiredCapabilities;
+ }
+
public List<Symbol> getOfferedCapabilities() {
return offeredCapabilities;
}
public void setOfferedCapabilities(List<Symbol> offeredCapabilities) {
- if (offeredCapabilities != null) {
+ if (offeredCapabilities == null) {
offeredCapabilities = Collections.emptyList();
}
@@ -410,7 +426,7 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
}
public void setOfferedProperties(Map<Symbol, Object> offeredProperties) {
- if (offeredProperties != null) {
+ if (offeredProperties == null) {
offeredProperties = Collections.emptyMap();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpAnonymousRelayTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpAnonymousRelayTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpAnonymousRelayTest.java
new file mode 100644
index 0000000..1743624
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpAnonymousRelayTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.amqp;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpReceiver;
+import org.apache.activemq.transport.amqp.client.AmqpSender;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.junit.Test;
+
+public class AmqpAnonymousRelayTest extends AmqpClientTestSupport {
+
+ @Override
+ protected boolean isAutoCreateQueues() {
+ return false;
+ }
+
+ @Override
+ protected boolean isAutoCreateAddresses() {
+ return false;
+ }
+
+ @Test(timeout = 60000)
+ public void testSendMessageOnAnonymousRelayLinkUsingMessageTo() throws Exception {
+ AmqpClient client = createAmqpClient();
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ AmqpSession session = connection.createSession();
+
+ AmqpSender sender = session.createAnonymousSender();
+ AmqpMessage message = new AmqpMessage();
+
+ message.setAddress(getQueueName());
+ message.setMessageId("msg" + 1);
+ message.setText("Test-Message");
+
+ sender.send(message);
+ sender.close();
+
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
+ receiver.flow(1);
+ AmqpMessage received = receiver.receive(10, TimeUnit.SECONDS);
+ assertNotNull("Should have read message", received);
+ assertEquals("msg1", received.getMessageId());
+ received.accept();
+
+ receiver.close();
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testSendMessageFailsOnAnonymousRelayLinkWhenNoToValueSet() throws Exception {
+ AmqpClient client = createAmqpClient();
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ AmqpSession session = connection.createSession();
+
+ AmqpSender sender = session.createAnonymousSender();
+ AmqpMessage message = new AmqpMessage();
+
+ message.setMessageId("msg" + 1);
+ message.setText("Test-Message");
+
+ try {
+ sender.send(message);
+ fail("Should not be able to send, message should be rejected");
+ } catch (Exception ex) {
+ ex.printStackTrace();
+ } finally {
+ sender.close();
+ }
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testSendMessageFailsOnAnonymousRelayWhenToFieldHasNonExistingAddress() throws Exception {
+ AmqpClient client = createAmqpClient();
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ AmqpSession session = connection.createSession();
+
+ AmqpSender sender = session.createAnonymousSender();
+ AmqpMessage message = new AmqpMessage();
+
+ message.setAddress("exampleQueu-not-in-service");
+ message.setMessageId("msg" + 1);
+ message.setText("Test-Message");
+
+ try {
+ sender.send(message);
+ fail("Should not be able to send, message should be rejected");
+ } catch (Exception ex) {
+ ex.printStackTrace();
+ } finally {
+ sender.close();
+ }
+ } finally {
+ connection.close();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java
index 99ce4db..02b1b99 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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.
@@ -14,11 +14,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.activemq.artemis.tests.integration.amqp;
import java.util.Set;
+import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
@@ -26,7 +26,6 @@ import org.apache.activemq.artemis.core.config.CoreAddressConfiguration;
import org.apache.activemq.artemis.core.config.CoreQueueConfiguration;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.Queue;
-import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.jms.server.JMSServerManager;
import org.apache.activemq.artemis.jms.server.impl.JMSServerManagerImpl;
@@ -48,9 +47,9 @@ public class AmqpClientTestSupport extends AmqpTestSupport {
protected static Symbol SHARED = Symbol.getSymbol("shared");
protected static Symbol GLOBAL = Symbol.getSymbol("global");
-
protected JMSServerManager serverManager;
protected ActiveMQServer server;
+
@Before
@Override
public void setUp() throws Exception {
@@ -84,6 +83,22 @@ public class AmqpClientTestSupport extends AmqpTestSupport {
super.tearDown();
}
+ protected boolean isAutoCreateQueues() {
+ return true;
+ }
+
+ protected boolean isAutoCreateAddresses() {
+ return true;
+ }
+
+ protected String getDeadLetterAddress() {
+ return "ActiveMQ.DLQ";
+ }
+
+ protected int getPrecreatedQueueSize() {
+ return 10;
+ }
+
protected ActiveMQServer createServer() throws Exception {
ActiveMQServer server = createServer(true, true);
serverManager = new JMSServerManagerImpl(server);
@@ -91,21 +106,30 @@ public class AmqpClientTestSupport extends AmqpTestSupport {
// Address 1
CoreAddressConfiguration address = new CoreAddressConfiguration();
- address.setName(getTestName()).getRoutingTypes().add(RoutingType.ANYCAST);
+ address.setName(getQueueName()).getRoutingTypes().add(RoutingType.ANYCAST);
CoreQueueConfiguration queueConfig = new CoreQueueConfiguration();
- queueConfig.setName(getTestName()).setAddress(getTestName()).setRoutingType(RoutingType.ANYCAST);
+ queueConfig.setName(getQueueName()).setAddress(getQueueName()).setRoutingType(RoutingType.ANYCAST);
address.getQueueConfigurations().add(queueConfig);
serverConfig.addAddressConfiguration(address);
- // Address 2
- CoreAddressConfiguration address2 = new CoreAddressConfiguration();
- address2.setName(getTestName2()).getRoutingTypes().add(RoutingType.ANYCAST);
- CoreQueueConfiguration queueConfig2 = new CoreQueueConfiguration();
- queueConfig2.setName(getTestName2()).setAddress(getTestName2()).setRoutingType(RoutingType.ANYCAST);
- address2.getQueueConfigurations().add(queueConfig2);
- serverConfig.addAddressConfiguration(address2);
+ // Address 1....N
+ for (int i = 0; i < getPrecreatedQueueSize(); ++i) {
+ CoreAddressConfiguration address2 = new CoreAddressConfiguration();
+ address2.setName(getQueueName(i)).getRoutingTypes().add(RoutingType.ANYCAST);
+ CoreQueueConfiguration queueConfig2 = new CoreQueueConfiguration();
+ queueConfig2.setName(getQueueName(i)).setAddress(getQueueName(i)).setRoutingType(RoutingType.ANYCAST);
+ address2.getQueueConfigurations().add(queueConfig2);
+ serverConfig.addAddressConfiguration(address2);
+ }
+
+ // Address configuration
+ AddressSettings addressSettings = new AddressSettings();
+
+ addressSettings.setAutoCreateQueues(isAutoCreateQueues());
+ addressSettings.setAutoCreateAddresses(isAutoCreateQueues());
+ addressSettings.setDeadLetterAddress(new SimpleString(getDeadLetterAddress()));
- serverConfig.getAddressesSettings().put("#", new AddressSettings().setAutoCreateQueues(true).setAutoCreateAddresses(true).setDeadLetterAddress(new SimpleString("ActiveMQ.DLQ")));
+ serverConfig.getAddressesSettings().put("#", addressSettings);
serverConfig.setSecurityEnabled(false);
Set<TransportConfiguration> acceptors = serverConfig.getAcceptorConfigurations();
for (TransportConfiguration tc : acceptors) {
@@ -127,8 +151,12 @@ public class AmqpClientTestSupport extends AmqpTestSupport {
return getName();
}
- public String getTestName2() {
- return getName() + "2";
+ public String getQueueName() {
+ return getName();
+ }
+
+ public String getQueueName(int index) {
+ return getName() + "-" + index;
}
public AmqpClientTestSupport() {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDeliveryAnnotationsTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDeliveryAnnotationsTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDeliveryAnnotationsTest.java
index 93ff22b..61787dd 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDeliveryAnnotationsTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDeliveryAnnotationsTest.java
@@ -40,18 +40,18 @@ public class AmqpDeliveryAnnotationsTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setText("Test-Message");
- message.setDeliveryAnnotation(DELIVERY_ANNOTATION_NAME, getTestName());
+ message.setDeliveryAnnotation(DELIVERY_ANNOTATION_NAME, getQueueName());
sender.send(message);
receiver.flow(1);
- Queue queue = getProxyToQueue(getTestName());
+ Queue queue = getProxyToQueue(getQueueName());
assertEquals(1, queue.getMessageCount());
AmqpMessage received = receiver.receive(5, TimeUnit.SECONDS);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDescribedTypePayloadTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDescribedTypePayloadTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDescribedTypePayloadTest.java
index 138f3cc..4b9213b 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDescribedTypePayloadTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDescribedTypePayloadTest.java
@@ -51,16 +51,16 @@ public class AmqpDescribedTypePayloadTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setDescribedType(new AmqpNoLocalFilter());
sender.send(message);
sender.close();
- Queue queue = getProxyToQueue(getTestName());
+ Queue queue = getProxyToQueue(getQueueName());
assertEquals(1, queue.getMessageCount());
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage received = receiver.receive(5, TimeUnit.SECONDS);
assertNotNull(received);
@@ -77,14 +77,14 @@ public class AmqpDescribedTypePayloadTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setDescribedType(new AmqpNoLocalFilter());
sender.send(message);
sender.close();
connection.close();
- Queue queue = getProxyToQueue(getTestName());
+ Queue queue = getProxyToQueue(getQueueName());
assertEquals(1, queue.getMessageCount());
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
@@ -111,13 +111,13 @@ public class AmqpDescribedTypePayloadTest extends AmqpClientTestSupport {
AmqpSession session = connection.createSession();
// Send with AMQP client.
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setDescribedType(new AmqpNoLocalFilter());
sender.send(message);
sender.close();
- Queue queue = getProxyToQueue(getTestName());
+ Queue queue = getProxyToQueue(getQueueName());
assertEquals(1, queue.getMessageCount());
// Receive and resend with OpenWire JMS client
@@ -142,7 +142,7 @@ public class AmqpDescribedTypePayloadTest extends AmqpClientTestSupport {
assertEquals(1, queue.getMessageCount());
// Now lets receive it with AMQP and see that we get back what we expected.
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage returned = receiver.receive(5, TimeUnit.SECONDS);
assertNotNull(returned);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpExpiredMessageTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpExpiredMessageTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpExpiredMessageTest.java
index 047d88d..440de12 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpExpiredMessageTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpExpiredMessageTest.java
@@ -35,10 +35,10 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
// Get the Queue View early to avoid racing the delivery.
- final Queue queueView = getProxyToQueue(getTestName());
+ final Queue queueView = getProxyToQueue(getQueueName());
assertNotNull(queueView);
AmqpMessage message = new AmqpMessage();
@@ -50,7 +50,7 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
assertEquals(1, queueView.getMessageCount());
// Now try and get the message
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage received = receiver.receive(1, TimeUnit.SECONDS);
assertNull(received);
@@ -66,10 +66,10 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
// Get the Queue View early to avoid racing the delivery.
- final Queue queueView = getProxyToQueue(getTestName());
+ final Queue queueView = getProxyToQueue(getQueueName());
assertNotNull(queueView);
AmqpMessage message = new AmqpMessage();
@@ -81,7 +81,7 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
assertEquals(1, queueView.getMessageCount());
// Now try and get the message
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage received = receiver.receive(5, TimeUnit.SECONDS);
assertNotNull(received);
@@ -97,10 +97,10 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
// Get the Queue View early to avoid racing the delivery.
- final Queue queueView = getProxyToQueue(getTestName());
+ final Queue queueView = getProxyToQueue(getQueueName());
assertNotNull(queueView);
AmqpMessage message = new AmqpMessage();
@@ -114,7 +114,7 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
assertEquals(1, queueView.getMessageCount());
// Now try and get the message
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage received = receiver.receive(1, TimeUnit.SECONDS);
assertNull(received);
@@ -130,10 +130,10 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
// Get the Queue View early to avoid racing the delivery.
- final Queue queueView = getProxyToQueue(getTestName());
+ final Queue queueView = getProxyToQueue(getQueueName());
assertNotNull(queueView);
AmqpMessage message = new AmqpMessage();
@@ -149,7 +149,7 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
Thread.sleep(1000);
// Now try and get the message
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage received = receiver.receive(1, TimeUnit.SECONDS);
assertNull(received);
@@ -165,10 +165,10 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
// Get the Queue View early to avoid racing the delivery.
- final Queue queueView = getProxyToQueue(getTestName());
+ final Queue queueView = getProxyToQueue(getQueueName());
assertNotNull(queueView);
AmqpMessage message = new AmqpMessage();
@@ -184,7 +184,7 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
assertEquals(1, queueView.getMessageCount());
// Now try and get the message
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage received = receiver.receive(5, TimeUnit.SECONDS);
assertNotNull(received);
@@ -200,10 +200,10 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
// Get the Queue View early to avoid racing the delivery.
- final Queue queueView = getProxyToQueue(getTestName());
+ final Queue queueView = getProxyToQueue(getQueueName());
assertNotNull(queueView);
AmqpMessage message = new AmqpMessage();
@@ -215,7 +215,7 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
assertEquals(1, queueView.getMessageCount());
// Now try and get the message
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage received = receiver.receive(5, TimeUnit.SECONDS);
assertNotNull(received);
@@ -231,10 +231,10 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
// Get the Queue View early to avoid racing the delivery.
- final Queue queueView = getProxyToQueue(getTestName());
+ final Queue queueView = getProxyToQueue(getQueueName());
assertNotNull(queueView);
AmqpMessage message = new AmqpMessage();
@@ -248,7 +248,7 @@ public class AmqpExpiredMessageTest extends AmqpClientTestSupport {
assertEquals(1, queueView.getMessageCount());
// Now try and get the message
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage received = receiver.receive(1, TimeUnit.SECONDS);
assertNull(received);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFailoverEndpointDiscoveryTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFailoverEndpointDiscoveryTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFailoverEndpointDiscoveryTest.java
new file mode 100644
index 0000000..81c2885
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFailoverEndpointDiscoveryTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.amqp;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.jms.Connection;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.activemq.artemis.api.core.TransportConfiguration;
+import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
+import org.apache.activemq.artemis.tests.integration.cluster.failover.FailoverTestBase;
+import org.apache.qpid.jms.JmsConnectionFactory;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class AmqpFailoverEndpointDiscoveryTest extends FailoverTestBase {
+
+ // this will ensure that all tests in this class are run twice,
+ // once with "true" passed to the class' constructor and once with "false"
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection<?> getParameters() {
+
+ // these 3 are for comparison
+ return Arrays.asList(new Object[][]{{"NON_SSL", 0}
+ /*, {"SSL", 1} */ });
+ }
+
+ private final int protocol;
+
+ public AmqpFailoverEndpointDiscoveryTest(String name, int protocol) {
+ this.protocol = protocol;
+ }
+
+ @Override
+ protected TransportConfiguration getAcceptorTransportConfiguration(final boolean live) {
+ return getNettyAcceptorTransportConfig(live);
+ }
+
+ @Override
+ protected TransportConfiguration getConnectorTransportConfiguration(final boolean live) {
+ return getNettyConnectorTransportConfig(live);
+ }
+
+ @Test(timeout = 120000)
+ public void testFailoverListWithAMQP() throws Exception {
+ JmsConnectionFactory factory = getJmsConnectionFactory();
+ try (Connection connection = factory.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ javax.jms.Queue queue = session.createQueue(ADDRESS.toString());
+ MessageProducer producer = session.createProducer(queue);
+ producer.send(session.createTextMessage("hello before failover"));
+ liveServer.crash(true, true);
+ producer.send(session.createTextMessage("hello after failover"));
+ MessageConsumer consumer = session.createConsumer(queue);
+ connection.start();
+ TextMessage receive = (TextMessage) consumer.receive(5000);
+ Assert.assertNotNull(receive);
+ Assert.assertEquals("hello before failover", receive.getText());
+ receive = (TextMessage) consumer.receive(5000);
+ Assert.assertEquals("hello after failover", receive.getText());
+ Assert.assertNotNull(receive);
+ }
+ }
+
+ private JmsConnectionFactory getJmsConnectionFactory() {
+ if (protocol == 0) {
+ return new JmsConnectionFactory("failover:(amqp://localhost:61616)");
+ } else {
+ String keystore = this.getClass().getClassLoader().getResource("client-side-keystore.jks").getFile();
+ String truststore = this.getClass().getClassLoader().getResource("client-side-truststore.jks").getFile();
+ return new JmsConnectionFactory("failover:(amqps://localhost:61616?transport.keyStoreLocation=" + keystore + "&transport.keyStorePassword=secureexample&transport.trustStoreLocation=" + truststore + "&transport.trustStorePassword=secureexample&transport.verifyHost=false)");
+ }
+ }
+
+ private TransportConfiguration getNettyAcceptorTransportConfig(final boolean live) {
+ Map<String, Object> server1Params = new HashMap<>();
+ if (protocol == 1) {
+ server1Params.put(TransportConstants.SSL_ENABLED_PROP_NAME, "true");
+
+ server1Params.put(TransportConstants.KEYSTORE_PATH_PROP_NAME, "server-side-keystore.jks");
+ server1Params.put(TransportConstants.KEYSTORE_PASSWORD_PROP_NAME, "secureexample");
+ server1Params.put(TransportConstants.TRUSTSTORE_PATH_PROP_NAME, "server-side-truststore.jks");
+ server1Params.put(TransportConstants.TRUSTSTORE_PASSWORD_PROP_NAME, "secureexample");
+ }
+
+ if (live) {
+ return new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, server1Params);
+ }
+
+ server1Params.put(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME, org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_PORT + 1);
+
+ return new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, server1Params);
+ }
+
+ private TransportConfiguration getNettyConnectorTransportConfig(final boolean live) {
+ Map<String, Object> server1Params = new HashMap<>();
+ if (protocol == 1) {
+ server1Params.put(TransportConstants.SSL_ENABLED_PROP_NAME, "true");
+ server1Params.put(TransportConstants.SSL_ENABLED_PROP_NAME, true);
+ server1Params.put(TransportConstants.TRUSTSTORE_PATH_PROP_NAME, "client-side-truststore.jks");
+ server1Params.put(TransportConstants.TRUSTSTORE_PASSWORD_PROP_NAME, "secureexample");
+ server1Params.put(TransportConstants.KEYSTORE_PATH_PROP_NAME, "client-side-keystore.jks");
+ server1Params.put(TransportConstants.KEYSTORE_PASSWORD_PROP_NAME, "secureexample");
+ }
+ if (live) {
+ return new TransportConfiguration(NETTY_CONNECTOR_FACTORY, server1Params);
+ }
+ server1Params.put(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME, org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_PORT + 1);
+ return new TransportConfiguration(NETTY_CONNECTOR_FACTORY, server1Params);
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpInboundConnectionTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpInboundConnectionTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpInboundConnectionTest.java
new file mode 100644
index 0000000..79bdf59
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpInboundConnectionTest.java
@@ -0,0 +1,255 @@
+/*
+ * 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.amqp;
+
+import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.ANONYMOUS_RELAY;
+import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.CONNECTION_OPEN_FAILED;
+import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.CONTAINER_ID;
+import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.DELAYED_DELIVERY;
+import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.INVALID_FIELD;
+import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.PRODUCT;
+import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.VERSION;
+import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.contains;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.apache.activemq.artemis.utils.VersionLoader;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpValidator;
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.transport.AmqpError;
+import org.apache.qpid.proton.amqp.transport.ErrorCondition;
+import org.apache.qpid.proton.engine.Connection;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests for behaviors expected of the broker when clients connect to the broker
+ */
+public class AmqpInboundConnectionTest extends AmqpClientTestSupport {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AmqpInboundConnectionTest.class);
+
+ private static final String BROKER_NAME = "localhost";
+ private static final String PRODUCT_NAME = "apache-activemq-artemis";
+
+ @Test
+ public void testBrokerContainerId() throws Exception {
+ AmqpClient client = createAmqpClient();
+ assertNotNull(client);
+
+ client.setValidator(new AmqpValidator() {
+
+ @Override
+ public void inspectOpenedResource(Connection connection) {
+ if (!BROKER_NAME.equals(connection.getRemoteContainer())) {
+ markAsInvalid("Broker did not send the expected container ID");
+ }
+ }
+ });
+
+ AmqpConnection connection = addConnection(client.connect());
+ try {
+ assertNotNull(connection);
+ connection.getStateInspector().assertValid();
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Test
+ public void testBrokerConnectionProperties() throws Exception {
+ AmqpClient client = createAmqpClient();
+
+ client.setValidator(new AmqpValidator() {
+
+ @Override
+ public void inspectOpenedResource(Connection connection) {
+
+ Map<Symbol, Object> properties = connection.getRemoteProperties();
+ if (!properties.containsKey(PRODUCT)) {
+ markAsInvalid("Broker did not send a queue product name value");
+ return;
+ }
+
+ if (!properties.containsKey(VERSION)) {
+ markAsInvalid("Broker did not send a queue version value");
+ return;
+ }
+
+ if (!PRODUCT_NAME.equals(properties.get(PRODUCT))) {
+ markAsInvalid("Broker did not send a the expected product name");
+ return;
+ }
+
+ String brokerVersion = VersionLoader.getVersion().getFullVersion();
+ if (!brokerVersion.equals(properties.get(VERSION))) {
+ markAsInvalid("Broker did not send a the expected product version");
+ return;
+ }
+ }
+ });
+
+ AmqpConnection connection = addConnection(client.connect());
+ try {
+ assertNotNull(connection);
+ connection.getStateInspector().assertValid();
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testConnectionCarriesExpectedCapabilities() throws Exception {
+ AmqpClient client = createAmqpClient();
+ assertNotNull(client);
+
+ client.setValidator(new AmqpValidator() {
+
+ @Override
+ public void inspectOpenedResource(Connection connection) {
+
+ Symbol[] offered = connection.getRemoteOfferedCapabilities();
+
+ if (!contains(offered, ANONYMOUS_RELAY)) {
+ markAsInvalid("Broker did not indicate it support anonymous relay");
+ return;
+ }
+
+ if (!contains(offered, DELAYED_DELIVERY)) {
+ markAsInvalid("Broker did not indicate it support delayed message delivery");
+ return;
+ }
+ }
+ });
+
+ AmqpConnection connection = addConnection(client.connect());
+ try {
+ assertNotNull(connection);
+ connection.getStateInspector().assertValid();
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testCanConnectWithDifferentContainerIds() throws Exception {
+ AmqpClient client = createAmqpClient();
+ assertNotNull(client);
+
+ AmqpConnection connection1 = addConnection(client.createConnection());
+ AmqpConnection connection2 = addConnection(client.createConnection());
+
+ connection1.setContainerId(getTestName() + "-Client:1");
+ connection2.setContainerId(getTestName() + "-Client:2");
+
+ connection1.connect();
+ assertEquals(1, server.getConnectionCount());
+
+ connection2.connect();
+ assertEquals(2, server.getConnectionCount());
+
+ connection1.close();
+ assertEquals(1, server.getConnectionCount());
+
+ connection2.close();
+ assertEquals(0, server.getConnectionCount());
+ }
+
+ @Test(timeout = 60000)
+ public void testCannotConnectWithSameContainerId() throws Exception {
+ AmqpClient client = createAmqpClient();
+
+ List<Symbol> desiredCapabilities = new ArrayList<>(1);
+ desiredCapabilities.add(AmqpSupport.SOLE_CONNECTION_CAPABILITY);
+
+ assertNotNull(client);
+
+ AmqpConnection connection1 = addConnection(client.createConnection());
+ AmqpConnection connection2 = addConnection(client.createConnection());
+
+ connection1.setDesiredCapabilities(desiredCapabilities);
+ connection2.setDesiredCapabilities(desiredCapabilities);
+
+ connection1.setContainerId(getTestName());
+ connection2.setContainerId(getTestName());
+
+ connection1.connect();
+ assertEquals(1, server.getConnectionCount());
+
+ connection2.setStateInspector(new AmqpValidator() {
+
+ @Override
+ public void inspectOpenedResource(Connection connection) {
+ if (!connection.getRemoteProperties().containsKey(CONNECTION_OPEN_FAILED)) {
+ markAsInvalid("Broker did not set connection establishment failed property");
+ }
+ }
+
+ @Override
+ public void inspectClosedResource(Connection connection) {
+ ErrorCondition remoteError = connection.getRemoteCondition();
+ if (remoteError == null || remoteError.getCondition() == null) {
+ markAsInvalid("Broker did not add error condition for duplicate client ID");
+ } else {
+ if (!remoteError.getCondition().equals(AmqpError.INVALID_FIELD)) {
+ markAsInvalid("Broker did not set condition to " + AmqpError.INVALID_FIELD);
+ }
+
+ if (!remoteError.getCondition().equals(AmqpError.INVALID_FIELD)) {
+ markAsInvalid("Broker did not set condition to " + AmqpError.INVALID_FIELD);
+ }
+ }
+
+ // Validate the info map contains a hint that the container/client id was the
+ // problem
+ Map<?, ?> infoMap = remoteError.getInfo();
+ if (infoMap == null) {
+ markAsInvalid("Broker did not set an info map on condition");
+ } else if (!infoMap.containsKey(INVALID_FIELD)) {
+ markAsInvalid("Info map does not contain expected key");
+ } else {
+ Object value = infoMap.get(INVALID_FIELD);
+ if (!CONTAINER_ID.equals(value)) {
+ markAsInvalid("Info map does not contain expected value: " + value);
+ }
+ }
+ }
+ });
+
+ try {
+ connection2.connect();
+ fail("Should not be able to connect with same container Id.");
+ } catch (Exception ex) {
+ LOG.info("Second connection with same container Id failed as expected.");
+ }
+
+ connection2.getStateInspector().assertValid();
+ connection2.close();
+
+ assertTrue(Wait.waitFor(() -> server.getConnectionCount() == 1));
+
+ connection1.close();
+ assertEquals(0, server.getConnectionCount());
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpManagementTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpManagementTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpManagementTest.java
new file mode 100644
index 0000000..c84a590
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpManagementTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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.amqp;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.artemis.api.core.management.ResourceNames;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpReceiver;
+import org.apache.activemq.transport.amqp.client.AmqpSender;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.apache.qpid.proton.amqp.messaging.AmqpValue;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class AmqpManagementTest extends AmqpClientTestSupport {
+
+ @Test
+ public void testManagementQueryOverAMQP() throws Throwable {
+ AmqpClient client = createAmqpClient();
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ String destinationAddress = getQueueName(1);
+ AmqpSession session = connection.createSession();
+ AmqpSender sender = session.createSender("activemq.management");
+ AmqpReceiver receiver = session.createReceiver(destinationAddress);
+ receiver.flow(10);
+
+ // Create request message for getQueueNames query
+ AmqpMessage request = new AmqpMessage();
+ request.setApplicationProperty("_AMQ_ResourceName", ResourceNames.BROKER);
+ request.setApplicationProperty("_AMQ_OperationName", "getQueueNames");
+ request.setReplyToAddress(destinationAddress);
+ request.setText("[]");
+
+ sender.send(request);
+ AmqpMessage response = receiver.receive(5, TimeUnit.SECONDS);
+ Assert.assertNotNull(response);
+ assertNotNull(response);
+ Object section = response.getWrappedMessage().getBody();
+ assertTrue(section instanceof AmqpValue);
+ Object value = ((AmqpValue) section).getValue();
+ assertTrue(value instanceof String);
+ assertTrue(((String) value).length() > 0);
+ assertTrue(((String) value).contains(destinationAddress));
+ response.accept();
+ } finally {
+ connection.close();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpMessagePriorityTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpMessagePriorityTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpMessagePriorityTest.java
index 2b57354..097d9bf 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpMessagePriorityTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpMessagePriorityTest.java
@@ -42,7 +42,7 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setMessageId("MessageID:1");
@@ -51,9 +51,9 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
sender.send(message);
sender.close();
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
- Queue queueView = getProxyToQueue(getTestName());
+ Queue queueView = getProxyToQueue(getQueueName());
assertEquals(1, queueView.getMessageCount());
receiver.flow(1);
@@ -73,7 +73,7 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setDurable(true);
@@ -91,9 +91,9 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
connection = addConnection(client.connect());
session = connection.createSession();
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
- Queue queueView = getProxyToQueue(getTestName());
+ Queue queueView = getProxyToQueue(getQueueName());
assertEquals(1, queueView.getMessageCount());
receiver.flow(1);
@@ -114,7 +114,7 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setMessageId("MessageID:1");
@@ -123,9 +123,9 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
sender.send(message);
sender.close();
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
- Queue queueView = getProxyToQueue(getTestName());
+ Queue queueView = getProxyToQueue(getQueueName());
assertEquals(1, queueView.getMessageCount());
receiver.flow(1);
@@ -146,7 +146,7 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setMessageId("MessageID:1");
@@ -155,9 +155,9 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
sender.send(message);
sender.close();
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
- Queue queueView = getProxyToQueue(getTestName());
+ Queue queueView = getProxyToQueue(getQueueName());
assertEquals(1, queueView.getMessageCount());
receiver.flow(1);
@@ -178,7 +178,7 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setMessageId("MessageID:1");
@@ -186,9 +186,9 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
sender.send(message);
sender.close();
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
- Queue queueView = getProxyToQueue(getTestName());
+ Queue queueView = getProxyToQueue(getQueueName());
assertEquals(1, queueView.getMessageCount());
receiver.flow(1);
@@ -208,7 +208,7 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
for (short i = 0; i <= 9; ++i) {
AmqpMessage message = new AmqpMessage();
@@ -219,9 +219,9 @@ public class AmqpMessagePriorityTest extends AmqpClientTestSupport {
sender.close();
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
- Queue queueView = getProxyToQueue(getTestName());
+ Queue queueView = getProxyToQueue(getQueueName());
assertEquals(10, queueView.getMessageCount());
receiver.flow(10);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpNettyFailoverTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpNettyFailoverTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpNettyFailoverTest.java
deleted file mode 100644
index 16cd7c3..0000000
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpNettyFailoverTest.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
- * 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.amqp;
-
-import org.apache.activemq.artemis.api.core.TransportConfiguration;
-import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
-import org.apache.activemq.artemis.tests.integration.cluster.failover.FailoverTestBase;
-import org.apache.qpid.jms.JmsConnectionFactory;
-import org.junit.Assert;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-import javax.jms.Connection;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-
-@RunWith(Parameterized.class)
-public class AmqpNettyFailoverTest extends FailoverTestBase {
-
-
- // this will ensure that all tests in this class are run twice,
- // once with "true" passed to the class' constructor and once with "false"
- @Parameterized.Parameters(name = "{0}")
- public static Collection getParameters() {
-
- // these 3 are for comparison
- return Arrays.asList(new Object[][]{{"NON_SSL", 0}
- /*, {"SSL", 1} */ });
- }
-
-
- private final int protocol;
-
- public AmqpNettyFailoverTest(String name, int protocol) {
- this.protocol = protocol;
- }
-
- @Override
- protected TransportConfiguration getAcceptorTransportConfiguration(final boolean live) {
- return getNettyAcceptorTransportConfig(live);
- }
-
- @Override
- protected TransportConfiguration getConnectorTransportConfiguration(final boolean live) {
- return getNettyConnectorTransportConfig(live);
- }
-
-
- @Test(timeout = 120000)
- public void testFailoverListWithAMQP() throws Exception {
- JmsConnectionFactory factory = getJmsConnectionFactory();
- try (Connection connection = factory.createConnection()) {
- Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- javax.jms.Queue queue = session.createQueue(ADDRESS.toString());
- MessageProducer producer = session.createProducer(queue);
- producer.send(session.createTextMessage("hello before failover"));
- liveServer.crash(true, true);
- producer.send(session.createTextMessage("hello after failover"));
- MessageConsumer consumer = session.createConsumer(queue);
- connection.start();
- TextMessage receive = (TextMessage) consumer.receive(5000);
- Assert.assertNotNull(receive);
- Assert.assertEquals("hello before failover", receive.getText());
- receive = (TextMessage) consumer.receive(5000);
- Assert.assertEquals("hello after failover", receive.getText());
- Assert.assertNotNull(receive);
- }
- }
-
- private JmsConnectionFactory getJmsConnectionFactory() {
- if (protocol == 0) {
- return new JmsConnectionFactory("failover:(amqp://localhost:61616)");
- } else {
- String keystore = this.getClass().getClassLoader().getResource("client-side-keystore.jks").getFile();
- String truststore = this.getClass().getClassLoader().getResource("client-side-truststore.jks").getFile();
- // return new JmsConnectionFactory("amqps://localhost:61616?transport.keyStoreLocation=" + keystore + "&transport.keyStorePassword=secureexample&transport.trustStoreLocation=" + truststore + "&transport.trustStorePassword=secureexample&transport.verifyHost=false");
- return new JmsConnectionFactory("failover:(amqps://localhost:61616?transport.keyStoreLocation=" + keystore + "&transport.keyStorePassword=secureexample&transport.trustStoreLocation=" + truststore + "&transport.trustStorePassword=secureexample&transport.verifyHost=false)");
- }
- }
-
- private TransportConfiguration getNettyAcceptorTransportConfig(final boolean live) {
- Map<String, Object> server1Params = new HashMap<>();
- if (protocol == 1) {
- server1Params.put(TransportConstants.SSL_ENABLED_PROP_NAME, "true");
-
- server1Params.put(TransportConstants.KEYSTORE_PATH_PROP_NAME, "server-side-keystore.jks");
- server1Params.put(TransportConstants.KEYSTORE_PASSWORD_PROP_NAME, "secureexample");
- server1Params.put(TransportConstants.TRUSTSTORE_PATH_PROP_NAME, "server-side-truststore.jks");
- server1Params.put(TransportConstants.TRUSTSTORE_PASSWORD_PROP_NAME, "secureexample");
- //server1Params.put(TransportConstants.NEED_CLIENT_AUTH_PROP_NAME, true);
-
- }
- if (live) {
- return new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, server1Params);
- }
-
-
- server1Params.put(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME, org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_PORT + 1);
-
- return new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, server1Params);
- }
-
- private TransportConfiguration getNettyConnectorTransportConfig(final boolean live) {
- Map<String, Object> server1Params = new HashMap<>();
- if (protocol == 1) {
- server1Params.put(TransportConstants.SSL_ENABLED_PROP_NAME, "true");
- server1Params.put(TransportConstants.SSL_ENABLED_PROP_NAME, true);
- server1Params.put(TransportConstants.TRUSTSTORE_PATH_PROP_NAME, "client-side-truststore.jks");
- server1Params.put(TransportConstants.TRUSTSTORE_PASSWORD_PROP_NAME, "secureexample");
- server1Params.put(TransportConstants.KEYSTORE_PATH_PROP_NAME, "client-side-keystore.jks");
- server1Params.put(TransportConstants.KEYSTORE_PASSWORD_PROP_NAME, "secureexample");
- }
- if (live) {
- return new TransportConfiguration(NETTY_CONNECTOR_FACTORY, server1Params);
- }
- server1Params.put(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME, org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_PORT + 1);
- return new TransportConfiguration(NETTY_CONNECTOR_FACTORY, server1Params);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpPresettledReceiverTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpPresettledReceiverTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpPresettledReceiverTest.java
index 657aff7..422e23e 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpPresettledReceiverTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpPresettledReceiverTest.java
@@ -35,16 +35,16 @@ public class AmqpPresettledReceiverTest extends AmqpClientTestSupport {
@Test(timeout = 60000)
public void testPresettledReceiverAndNonPresettledReceiverOnSameQueue() throws Exception {
final int MSG_COUNT = 2;
- sendMessages(getTestName(), MSG_COUNT);
+ sendMessages(getQueueName(), MSG_COUNT);
AmqpClient client = createAmqpClient();
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpReceiver receiver1 = session.createReceiver(getTestName(), null, false, true);
- AmqpReceiver receiver2 = session.createReceiver(getTestName());
+ AmqpReceiver receiver1 = session.createReceiver(getQueueName(), null, false, true);
+ AmqpReceiver receiver2 = session.createReceiver(getQueueName());
- final Queue queueView = getProxyToQueue(getTestName());
+ final Queue queueView = getProxyToQueue(getQueueName());
assertEquals(MSG_COUNT, queueView.getMessageCount());
receiver1.flow(1);
@@ -68,7 +68,7 @@ public class AmqpPresettledReceiverTest extends AmqpClientTestSupport {
System.out.println("Message Count after all consumed: " + queueView.getMessageCount());
// Should be nothing left on the Queue
- AmqpReceiver receiver3 = session.createReceiver(getTestName());
+ AmqpReceiver receiver3 = session.createReceiver(getQueueName());
receiver3.flow(1);
AmqpMessage received = receiver3.receive(5, TimeUnit.SECONDS);
@@ -85,15 +85,15 @@ public class AmqpPresettledReceiverTest extends AmqpClientTestSupport {
@Test(timeout = 60000)
public void testPresettledReceiverReadsAllMessages() throws Exception {
final int MSG_COUNT = 100;
- sendMessages(getTestName(), MSG_COUNT);
+ sendMessages(getQueueName(), MSG_COUNT);
AmqpClient client = createAmqpClient();
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpReceiver receiver = session.createReceiver(getTestName(), null, false, true);
+ AmqpReceiver receiver = session.createReceiver(getQueueName(), null, false, true);
- final Queue queueView = getProxyToQueue(getTestName());
+ final Queue queueView = getProxyToQueue(getQueueName());
assertEquals(MSG_COUNT, queueView.getMessageCount());
receiver.flow(MSG_COUNT);
@@ -105,7 +105,7 @@ public class AmqpPresettledReceiverTest extends AmqpClientTestSupport {
System.out.println("Message Count after all consumed: " + queueView.getMessageCount());
// Open a new receiver and see if any message are left on the Queue
- receiver = session.createReceiver(getTestName());
+ receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage received = receiver.receive(5, TimeUnit.SECONDS);
if (received != null) {
@@ -121,15 +121,15 @@ public class AmqpPresettledReceiverTest extends AmqpClientTestSupport {
@Test(timeout = 60000)
public void testPresettledReceiverReadsAllMessagesInWhenReadInBatches() throws Exception {
final int MSG_COUNT = 100;
- sendMessages(getTestName(), MSG_COUNT);
+ sendMessages(getQueueName(), MSG_COUNT);
AmqpClient client = createAmqpClient();
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpReceiver receiver = session.createReceiver(getTestName(), null, false, true);
+ AmqpReceiver receiver = session.createReceiver(getQueueName(), null, false, true);
- final Queue queueView = getProxyToQueue(getTestName());
+ final Queue queueView = getProxyToQueue(getQueueName());
assertEquals(MSG_COUNT, queueView.getMessageCount());
// Consume all 100 but do so in batches by flowing only limited credit.
@@ -157,7 +157,7 @@ public class AmqpPresettledReceiverTest extends AmqpClientTestSupport {
System.out.println("Message Count after all consumed: " + queueView.getMessageCount());
// Open a new receiver and see if any message are left on the Queue
- receiver = session.createReceiver(getTestName());
+ receiver = session.createReceiver(getQueueName());
receiver.flow(1);
AmqpMessage received = receiver.receive(5, TimeUnit.SECONDS);
if (received != null) {
@@ -185,8 +185,8 @@ public class AmqpPresettledReceiverTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
- final Queue queue = getProxyToQueue(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
+ final Queue queue = getProxyToQueue(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setText("Test-Message");
@@ -194,7 +194,7 @@ public class AmqpPresettledReceiverTest extends AmqpClientTestSupport {
assertEquals(1, queue.getMessageCount());
- AmqpReceiver receiver = session.createReceiver(getTestName(), null, false, true);
+ AmqpReceiver receiver = session.createReceiver(getQueueName(), null, false, true);
session.begin();
@@ -221,8 +221,8 @@ public class AmqpPresettledReceiverTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
- final Queue queue = getProxyToQueue(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
+ final Queue queue = getProxyToQueue(getQueueName());
AmqpMessage message = new AmqpMessage();
message.setText("Test-Message");
@@ -230,7 +230,7 @@ public class AmqpPresettledReceiverTest extends AmqpClientTestSupport {
assertEquals(1, queue.getMessageCount());
- AmqpReceiver receiver = session.createReceiver(getTestName(), null, false, true);
+ AmqpReceiver receiver = session.createReceiver(getQueueName(), null, false, true);
session.begin();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDispositionTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDispositionTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDispositionTest.java
index b347d37..9cd8f50 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDispositionTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDispositionTest.java
@@ -34,19 +34,18 @@ public class AmqpReceiverDispositionTest extends AmqpClientTestSupport {
@Test(timeout = 30000)
public void testReleasedDisposition() throws Exception {
- sendMessages(getTestName(), 1);
+ sendMessages(getQueueName(), 1);
AmqpClient client = createAmqpClient();
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpReceiver receiver1 = session.createReceiver(getTestName());
+ AmqpReceiver receiver1 = session.createReceiver(getQueueName());
receiver1.flow(1);
AmqpMessage message = receiver1.receive(5, TimeUnit.SECONDS);
- AmqpReceiver receiver2 = session.createReceiver(getTestName());
-
+ AmqpReceiver receiver2 = session.createReceiver(getQueueName());
assertNotNull("did not receive message first time", message);
assertEquals("MessageID:0", message.getMessageId());
@@ -75,13 +74,13 @@ public class AmqpReceiverDispositionTest extends AmqpClientTestSupport {
@Test(timeout = 30000)
public void testRejectedDisposition() throws Exception {
- sendMessages(getTestName(), 1);
+ sendMessages(getQueueName(), 1);
AmqpClient client = createAmqpClient();
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpReceiver receiver1 = session.createReceiver(getTestName());
+ AmqpReceiver receiver1 = session.createReceiver(getQueueName());
receiver1.flow(1);
AmqpMessage message = receiver1.receive(5, TimeUnit.SECONDS);
@@ -101,7 +100,7 @@ public class AmqpReceiverDispositionTest extends AmqpClientTestSupport {
assertNull("Should not receive message again", message);
// Attempt to Read the message again with another receiver to validate it is archived.
- AmqpReceiver receiver2 = session.createReceiver(getTestName());
+ AmqpReceiver receiver2 = session.createReceiver(getQueueName());
receiver2.flow(1);
assertNull(receiver2.receiveNoWait());
@@ -129,13 +128,13 @@ public class AmqpReceiverDispositionTest extends AmqpClientTestSupport {
}
private void doModifiedDispositionTestImpl(Boolean deliveryFailed, Boolean undeliverableHere) throws Exception {
- sendMessages(getTestName(), 1);
+ sendMessages(getQueueName(), 1);
AmqpClient client = createAmqpClient();
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- AmqpReceiver receiver1 = session.createReceiver(getTestName());
+ AmqpReceiver receiver1 = session.createReceiver(getQueueName());
receiver1.flow(1);
AmqpMessage message = receiver1.receive(5, TimeUnit.SECONDS);
@@ -154,7 +153,7 @@ public class AmqpReceiverDispositionTest extends AmqpClientTestSupport {
assertNull("Should not receive message again", message);
}
- AmqpReceiver receiver2 = session.createReceiver(getTestName());
+ AmqpReceiver receiver2 = session.createReceiver(getQueueName());
receiver2.flow(1);
message = receiver2.receive(5, TimeUnit.SECONDS);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDrainTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDrainTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDrainTest.java
index 1af9028..681ffbd 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDrainTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDrainTest.java
@@ -35,15 +35,15 @@ public class AmqpReceiverDrainTest extends AmqpClientTestSupport {
@Test(timeout = 60000)
public void testReceiverCanDrainMessages() throws Exception {
int MSG_COUNT = 20;
- sendMessages(getTestName(), MSG_COUNT);
+ sendMessages(getQueueName(), MSG_COUNT);
AmqpClient client = createAmqpClient();
AmqpConnection connection = client.connect();
AmqpSession session = connection.createSession();
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
- Queue queueView = getProxyToQueue(getTestName());
+ Queue queueView = getProxyToQueue(getQueueName());
assertEquals(MSG_COUNT, queueView.getMessageCount());
receiver.drain(MSG_COUNT);
@@ -66,11 +66,11 @@ public class AmqpReceiverDrainTest extends AmqpClientTestSupport {
AmqpConnection connection = client.connect();
AmqpSession session = connection.createSession();
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(10);
- Queue queueView = getProxyToQueue(getTestName());
+ Queue queueView = getProxyToQueue(getQueueName());
assertEquals(0, queueView.getMessageCount());
assertEquals(0, queueView.getDeliveringCount());
@@ -86,15 +86,15 @@ public class AmqpReceiverDrainTest extends AmqpClientTestSupport {
@Test(timeout = 60000)
public void testPullOneFromRemote() throws Exception {
int MSG_COUNT = 20;
- sendMessages(getTestName(), MSG_COUNT);
+ sendMessages(getQueueName(), MSG_COUNT);
AmqpClient client = createAmqpClient();
AmqpConnection connection = client.connect();
AmqpSession session = connection.createSession();
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
- Queue queueView = getProxyToQueue(getTestName());
+ Queue queueView = getProxyToQueue(getQueueName());
assertEquals(MSG_COUNT, queueView.getMessageCount());
assertEquals(0, receiver.getReceiver().getRemoteCredit());
@@ -119,11 +119,11 @@ public class AmqpReceiverDrainTest extends AmqpClientTestSupport {
AmqpConnection connection = client.connect();
AmqpSession session = connection.createSession();
- AmqpReceiver receiver = session.createReceiver(getTestName());
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
receiver.flow(10);
- Queue queueView = getProxyToQueue(getTestName());
+ Queue queueView = getProxyToQueue(getQueueName());
assertEquals(0, queueView.getMessageCount());
assertEquals(10, receiver.getReceiver().getRemoteCredit());
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0260a304/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverWithFiltersTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverWithFiltersTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverWithFiltersTest.java
index f8f726a..1e140c7 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverWithFiltersTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverWithFiltersTest.java
@@ -74,7 +74,7 @@ public class AmqpReceiverWithFiltersTest extends AmqpClientTestSupport {
filters.put(AmqpUnknownFilterType.UNKNOWN_FILTER_NAME, AmqpUnknownFilterType.UNKNOWN_FILTER);
Source source = new Source();
- source.setAddress(getTestName());
+ source.setAddress(getQueueName());
source.setFilter(filters);
source.setDurable(TerminusDurability.NONE);
source.setExpiryPolicy(TerminusExpiryPolicy.LINK_DETACH);
@@ -116,13 +116,12 @@ public class AmqpReceiverWithFiltersTest extends AmqpClientTestSupport {
AmqpConnection connection = addConnection(client.connect());
AmqpSession session = connection.createSession();
- session.createReceiver(getTestName(), "color = red");
+ session.createReceiver(getQueueName(), "color = red");
connection.getStateInspector().assertValid();
connection.close();
}
-
@Test(timeout = 60000)
public void testReceivedUnsignedFilter() throws Exception {
final int NUM_MESSAGES = 100;
@@ -131,10 +130,9 @@ public class AmqpReceiverWithFiltersTest extends AmqpClientTestSupport {
AmqpConnection connection = client.connect();
try {
-
// Normal Session which won't create an TXN itself
AmqpSession session = connection.createSession();
- AmqpSender sender = session.createSender(getTestName());
+ AmqpSender sender = session.createSender(getQueueName());
for (int i = 0; i < NUM_MESSAGES + 1; ++i) {
AmqpMessage message = new AmqpMessage();
@@ -144,7 +142,7 @@ public class AmqpReceiverWithFiltersTest extends AmqpClientTestSupport {
}
// Read all messages from the Queue, do not accept them yet.
- AmqpReceiver receiver = session.createReceiver(getTestName(), "myNewID < " + (NUM_MESSAGES / 2));
+ AmqpReceiver receiver = session.createReceiver(getQueueName(), "myNewID < " + (NUM_MESSAGES / 2));
ArrayList<AmqpMessage> messages = new ArrayList<>(NUM_MESSAGES);
receiver.flow((NUM_MESSAGES + 2) * 2);
for (int i = 0; i < NUM_MESSAGES / 2; ++i) {
@@ -161,6 +159,4 @@ public class AmqpReceiverWithFiltersTest extends AmqpClientTestSupport {
connection.close();
}
}
-
-
}