This is an automated email from the ASF dual-hosted git repository.
michaelpearce pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 0f905224e7cd9f71729ebd8434cdde44e396fd0c
Author: Clebert Suconic <clebertsuco...@apache.org>
AuthorDate: Thu Jan 24 16:01:21 2019 -0500
ARTEMIS-2238 Enhancement to queueQuery on producer
---
.../jms/client/ActiveMQMessageProducer.java | 105 +++++++++++-------
.../client/AutoCreateJmsDestinationTest.java | 122 +++++++++++++++++++++
2 files changed, 188 insertions(+), 39 deletions(-)
diff --git
a/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQMessageProducer.java
b/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQMessageProducer.java
index b6c72a7..a813724 100644
---
a/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQMessageProducer.java
+++
b/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQMessageProducer.java
@@ -47,12 +47,15 @@ import
org.apache.activemq.artemis.api.core.client.SendAcknowledgementHandler;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.utils.UUID;
import org.apache.activemq.artemis.utils.UUIDGenerator;
+import org.jboss.logging.Logger;
/**
* ActiveMQ Artemis implementation of a JMS MessageProducer.
*/
public class ActiveMQMessageProducer implements MessageProducer, QueueSender,
TopicPublisher {
+ private static final Logger logger =
Logger.getLogger(ActiveMQMessageProducer.class);
+
private final ConnectionFactoryOptions options;
private final ActiveMQConnection connection;
@@ -403,47 +406,10 @@ public class ActiveMQMessageProducer implements
MessageProducer, QueueSender, To
}
address = destination.getSimpleAddress();
-
- if (!connection.containsKnownDestination(address)) {
- try {
- ClientSession.AddressQuery query =
clientSession.addressQuery(address);
-
- if (!query.isExists()) {
- if (destination.isQueue() && query.isAutoCreateQueues()) {
- clientSession.createAddress(address, RoutingType.ANYCAST,
true);
- if (destination.isTemporary()) {
- // TODO is it right to use the address for the queue
name here?
- session.createTemporaryQueue(destination,
RoutingType.ANYCAST, address, null, query);
- } else {
- session.createQueue(destination, RoutingType.ANYCAST,
address, null, true, true, query);
- }
- } else if (!destination.isQueue() &&
query.isAutoCreateAddresses()) {
- clientSession.createAddress(address,
RoutingType.MULTICAST, true);
- } else if ((destination.isQueue() &&
!query.isAutoCreateQueues()) || (!destination.isQueue() &&
!query.isAutoCreateAddresses())) {
- throw new InvalidDestinationException("Destination " +
address + " does not exist");
- }
- } else {
- if (destination.isQueue()) {
- ClientSession.QueueQuery queueQuery =
clientSession.queueQuery(address);
- if (!queueQuery.isExists()) {
- if (destination.isTemporary()) {
- session.createTemporaryQueue(destination,
RoutingType.ANYCAST, address, null, query);
- } else {
- session.createQueue(destination,
RoutingType.ANYCAST, address, null, true, true, query);
- }
- }
- }
-
- connection.addKnownDestination(address);
- }
- } catch (ActiveMQQueueExistsException e) {
- // The queue was created by another client/admin between the
query check and send create queue packet
- } catch (ActiveMQException e) {
- throw JMSExceptionHelper.convertFromActiveMQException(e);
- }
- }
}
+ checkDestination(destination, address, clientSession);
+
ActiveMQMessage activeMQJmsMessage;
boolean foreign = false;
@@ -533,6 +499,67 @@ public class ActiveMQMessageProducer implements
MessageProducer, QueueSender, To
}
}
+ private void checkDestination(ActiveMQDestination destination,
+ SimpleString address,
+ ClientSession clientSession) throws
JMSException {
+
+ // TODO: What to do with FQQN
+ if (!connection.containsKnownDestination(address)) {
+ try {
+ ClientSession.AddressQuery addressQuery =
clientSession.addressQuery(address);
+
+ boolean addressExists = addressQuery.isExists();
+ // first we check the address existence, and autoCreate it if
allowed in case it does not exists
+
+ if (!addressExists && addressQuery.isAutoCreateAddresses()) {
+
+ if (destination.isQueue() &&
!addressQuery.isAutoCreateQueues()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Address " + address + " was not created
because we would not have permission to create queue");
+ }
+ // if it can't create the internal queue on JMS Queues, why
bother creating the address, just mark it false now
+ addressExists = false;
+ } else {
+ RoutingType addressType = destination.isQueue() ?
RoutingType.ANYCAST : RoutingType.MULTICAST;
+ clientSession.createAddress(address, addressType, true);
+ addressExists = true;
+ }
+ }
+
+
+ // Second we create the queue, but we only do it if the address
was created
+ if (destination.isQueue() && addressExists) {
+ ClientSession.QueueQuery queueQuery =
clientSession.queueQuery(address);
+ if (!queueQuery.isExists()) {
+ if (addressQuery.isAutoCreateQueues()) {
+ try {
+ if (destination.isTemporary()) {
+ session.createTemporaryQueue(destination,
RoutingType.ANYCAST, address, null, addressQuery);
+ } else {
+ session.createQueue(destination,
RoutingType.ANYCAST, address, null, true, true, addressQuery);
+ }
+ } catch (ActiveMQQueueExistsException thatsOK) {
+ // nothing to be done
+ }
+ } else {
+ throw new InvalidDestinationException("Queue " + address
+ " does not exist");
+ }
+ }
+ }
+
+ if (!addressExists) {
+ throw new InvalidDestinationException("Address " + address + "
does not exist");
+ }
+
+
+ // this is done at the end, if no exceptions are thrown
+ connection.addKnownDestination(address);
+ } catch (ActiveMQException e) {
+ throw JMSExceptionHelper.convertFromActiveMQException(e);
+ }
+ }
+ }
+
private void checkClosed() throws JMSException {
if (clientProducer.isClosed()) {
throw new IllegalStateException("Producer is closed");
diff --git
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AutoCreateJmsDestinationTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AutoCreateJmsDestinationTest.java
index 3094c59..a7a7a87 100644
---
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AutoCreateJmsDestinationTest.java
+++
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AutoCreateJmsDestinationTest.java
@@ -17,26 +17,42 @@
package org.apache.activemq.artemis.tests.integration.client;
import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
import javax.jms.JMSSecurityException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
+import javax.jms.Topic;
import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.api.core.client.ClientSession;
+import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
+import org.apache.activemq.artemis.api.core.client.ServerLocator;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
+import org.apache.activemq.artemis.core.client.impl.ServerLocatorImpl;
import org.apache.activemq.artemis.core.security.Role;
import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.jms.client.ActiveMQConnection;
+import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.apache.activemq.artemis.jms.client.ActiveMQQueue;
import org.apache.activemq.artemis.jms.client.ActiveMQTemporaryTopic;
+import org.apache.activemq.artemis.jms.client.ActiveMQTopic;
import org.apache.activemq.artemis.junit.Wait;
import
org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
import org.apache.activemq.artemis.tests.integration.IntegrationTestLogger;
import org.apache.activemq.artemis.tests.util.JMSTestBase;
+import org.apache.activemq.artemis.utils.UUIDGenerator;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -48,6 +64,9 @@ public class AutoCreateJmsDestinationTest extends JMSTestBase
{
public static final String QUEUE_NAME = "test";
+ ClientSessionFactory factory;
+ ClientSession clientSession;
+
@Test
public void testAutoCreateOnSendToQueue() throws Exception {
Connection connection = cf.createConnection();
@@ -265,6 +284,98 @@ public class AutoCreateJmsDestinationTest extends
JMSTestBase {
connection.close();
}
+
+ @Test //(timeout = 30000)
+ // QueueAutoCreationTest was created to validate auto-creation of queues
+ // and this test was added to validate a regression:
https://issues.apache.org/jira/browse/ARTEMIS-2238
+ public void testAutoCreateOnTopic() throws Exception {
+ ConnectionFactory factory = new ActiveMQConnectionFactory();
+ Connection connection = factory.createConnection();
+ SimpleString addressName =
UUIDGenerator.getInstance().generateSimpleStringUUID();
+ System.out.println("Address is " + addressName);
+ clientSession.createAddress(addressName, RoutingType.ANYCAST, false);
+ Topic topic = new ActiveMQTopic(addressName.toString());
+ Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(topic);
+ for (int i = 0; i < 10; i++) {
+ producer.send(session.createTextMessage("hello"));
+ }
+
+
Assert.assertTrue(((ActiveMQConnection)connection).containsKnownDestination(addressName));
+ }
+
+ @Test (timeout = 30000)
+ // QueueAutoCreationTest was created to validate auto-creation of queues
+ // and this test was added to validate a regression:
https://issues.apache.org/jira/browse/ARTEMIS-2238
+ public void testAutoCreateOnAddressOnly() throws Exception {
+
+ server.getAddressSettingsRepository().clear();
+ AddressSettings settings = new
AddressSettings().setAutoCreateAddresses(true).setAutoCreateQueues(false);
+ server.getAddressSettingsRepository().addMatch("#", settings);
+
+ ConnectionFactory factory = new ActiveMQConnectionFactory();
+ try (Connection connection = factory.createConnection()) {
+ SimpleString addressName =
UUIDGenerator.getInstance().generateSimpleStringUUID();
+ System.out.println("Address is " + addressName);
+ javax.jms.Queue queue = new ActiveMQQueue(addressName.toString());
+ Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(null);
+ try {
+ producer.send(queue, session.createTextMessage("hello"));
+ Assert.fail("Expected to throw exception here");
+ } catch (JMSException expected) {
+ }
+
Assert.assertFalse(((ActiveMQConnection)connection).containsKnownDestination(addressName));
+ }
+
+ }
+
+ @Test (timeout = 30000)
+ // QueueAutoCreationTest was created to validate auto-creation of queues
+ // and this test was added to validate a regression:
https://issues.apache.org/jira/browse/ARTEMIS-2238
+ public void testAutoCreateOnAddressOnlyDuringProducerCreate() throws
Exception {
+ server.getAddressSettingsRepository().clear();
+ AddressSettings settings = new
AddressSettings().setAutoCreateAddresses(true).setAutoCreateQueues(false);
+ server.getAddressSettingsRepository().addMatch("#", settings);
+
+ ConnectionFactory factory = new ActiveMQConnectionFactory();
+ Connection connection = factory.createConnection();
+ SimpleString addressName =
UUIDGenerator.getInstance().generateSimpleStringUUID();
+ clientSession.createAddress(addressName, RoutingType.ANYCAST, true); //
this will force the system to create the address only
+ javax.jms.Queue queue = new ActiveMQQueue(addressName.toString());
+ Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ try {
+ MessageProducer producer = session.createProducer(queue);
+ Assert.fail("Exception expected");
+ } catch (JMSException expected) {
+ }
+
+
Assert.assertFalse(((ActiveMQConnection)connection).containsKnownDestination(addressName));
+ }
+
+
+ @Test (timeout = 30000)
+ // QueueAutoCreationTest was created to validate auto-creation of queues
+ // and this test was added to validate a regression:
https://issues.apache.org/jira/browse/ARTEMIS-2238
+ public void testAutoCreateOnAddressOnlyDuringProducerCreateQueueSucceed()
throws Exception {
+ server.getAddressSettingsRepository().clear();
+ AddressSettings settings = new
AddressSettings().setAutoCreateAddresses(true).setAutoCreateQueues(true);
+ server.getAddressSettingsRepository().addMatch("#", settings);
+
+ ConnectionFactory factory = cf;
+ try(Connection connection = factory.createConnection()) {
+ SimpleString addressName =
UUIDGenerator.getInstance().generateSimpleStringUUID();
+ clientSession.createAddress(addressName, RoutingType.ANYCAST, true);
// this will force the system to create the address only
+ javax.jms.Queue queue = new ActiveMQQueue(addressName.toString());
+ Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(queue);
+ Assert.assertNotNull(server.locateQueue(addressName));
+
+ Assert.assertTrue(((ActiveMQConnection)
connection).containsKnownDestination(addressName));
+ }
+ }
+
+
@Before
@Override
public void setUp() throws Exception {
@@ -276,6 +387,17 @@ public class AutoCreateJmsDestinationTest extends
JMSTestBase {
Set<Role> roles = new HashSet<>();
roles.add(role);
server.getSecurityRepository().addMatch("#", roles);
+ ServerLocator locator =
ServerLocatorImpl.newLocator("tcp://localhost:61616");
+ factory = locator.createSessionFactory();
+ clientSession = factory.createSession();
+ }
+
+ @After
+ @Override
+ public void tearDown() throws Exception {
+ clientSession.close();
+ factory.close();
+ super.tearDown();
}
@Override
