activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From clebertsuco...@apache.org
Subject [01/50] [abbrv] activemq-artemis git commit: ARTEMIS-876 Internalise Cluster Namespace and remove JMS Prefix
Date Fri, 09 Dec 2016 19:48:45 GMT
Repository: activemq-artemis
Updated Branches:
  refs/heads/master ebbc91c72 -> 6682072da


ARTEMIS-876 Internalise Cluster Namespace and remove JMS Prefix


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

Branch: refs/heads/master
Commit: 0006627d120a9010f85aed16d8678e0a61073585
Parents: 84e8a87
Author: Martyn Taylor <mtaylor@redhat.com>
Authored: Mon Nov 7 11:36:18 2016 +0000
Committer: Martyn Taylor <mtaylor@redhat.com>
Committed: Fri Dec 9 18:43:15 2016 +0000

----------------------------------------------------------------------
 .../artemis/cli/commands/etc/broker.xml         | 26 +++++---
 .../cli/commands/etc/cluster-settings.txt       |  1 -
 .../config/ActiveMQDefaultConfiguration.java    |  6 ++
 .../artemis/jms/client/ActiveMQSession.java     | 32 ++++++---
 .../artemis/core/config/Configuration.java      |  3 +
 .../core/config/impl/ConfigurationImpl.java     | 13 ++++
 .../deployers/impl/FileConfigurationParser.java |  4 ++
 .../artemis/core/server/ActiveMQServer.java     |  4 ++
 .../cluster/impl/ClusterConnectionBridge.java   | 22 +++++-
 .../cluster/impl/ClusterConnectionImpl.java     | 10 ++-
 .../core/server/impl/ActiveMQServerImpl.java    | 34 +++++++---
 .../core/server/impl/ScaleDownHandler.java      |  3 +-
 .../resources/schema/artemis-configuration.xsd  | 70 +++-----------------
 .../test/resources/artemis-configuration.xsd    |  2 +-
 .../crossprotocol/AMQPToOpenwireTest.java       | 13 ++--
 .../integration/mqtt/imported/MQTTTest.java     |  6 +-
 .../integration/server/ScaleDown3NodeTest.java  |  4 +-
 .../tests/integration/server/ScaleDownTest.java |  2 +-
 18 files changed, 148 insertions(+), 107 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
index 23f2109..0871ab6 100644
--- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
@@ -22,12 +22,8 @@ under the License.
                xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd">
 
-   <jms xmlns="urn:activemq:jms">
-      <queue name="DLQ"/>
-      <queue name="ExpiryQueue"/>${jms-list.settings}
-   </jms>
-
-   <core xmlns="urn:activemq:core">
+   <core xmlns="urn:activemq:core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="urn:activemq:core ">
 
       <name>${name}</name>
 
@@ -89,8 +85,8 @@ ${cluster-security.settings}${cluster.settings}${replicated.settings}${shared-st
       <address-settings>
          <!--default for catch all-->
          <address-setting match="#">
-            <dead-letter-address>jms.queue.DLQ</dead-letter-address>
-            <expiry-address>jms.queue.ExpiryQueue</expiry-address>
+            <dead-letter-address>DLQ</dead-letter-address>
+            <expiry-address>ExpiryQueue</expiry-address>
             <redelivery-delay>0</redelivery-delay>
             <!-- with -1 only the global-max-size is in use for limiting -->
             <max-size-bytes>-1</max-size-bytes>
@@ -98,5 +94,19 @@ ${cluster-security.settings}${cluster.settings}${replicated.settings}${shared-st
             <address-full-policy>PAGE</address-full-policy>
          </address-setting>
       </address-settings>
+
+      <addresses>
+         <address name="DLQ" type="anycast">
+            <queues>
+               <queue name="DLQ" />
+            </queues>
+         </address>
+         <address name="ExpiryQueue" type="anycast">
+            <queues>
+               <queue name="ExpiryQueue" />
+            </queues>
+         </address>
+      </addresses>
+
    </core>
 </configuration>

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/cluster-settings.txt
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/cluster-settings.txt
b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/cluster-settings.txt
index b7afbca..f639735 100644
--- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/cluster-settings.txt
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/cluster-settings.txt
@@ -18,7 +18,6 @@
 
       <cluster-connections>
          <cluster-connection name="my-cluster">
-            <address>jms</address>
             <connector-ref>artemis</connector-ref>
             <message-load-balancing>${message-load-balancing}</message-load-balancing>
             <max-hops>${max-hops}</max-hops>

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
index 9791634..e9212f2 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
@@ -453,6 +453,8 @@ public final class ActiveMQDefaultConfiguration {
 
    public static String DEFAULT_NETWORK_CHECK_NIC = null;
 
+   public static final String DEFAULT_INTERNAL_NAMING_PREFIX = "$.artemis.internal.";
+
    /**
     * If true then the ActiveMQ Artemis Server will make use of any Protocol Managers that
are in available on the classpath. If false then only the core protocol will be available,
unless in Embedded mode where users can inject their own Protocol Managers.
     */
@@ -1199,6 +1201,10 @@ public final class ActiveMQDefaultConfiguration {
       return DEFAULT_DELETE_QUEUE_ON_NO_CONSUMERS;
    }
 
+   public static String getInternalNamingPrefix() {
+      return DEFAULT_INTERNAL_NAMING_PREFIX;
+   }
+
    public static String getDefaultSystemPropertyPrefix() {
       return DEFAULT_SYSTEM_PROPERTY_PREFIX;
    }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQSession.java
----------------------------------------------------------------------
diff --git a/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQSession.java
b/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQSession.java
index d40ca21..d554cf8 100644
--- a/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQSession.java
+++ b/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQSession.java
@@ -299,19 +299,31 @@ public class ActiveMQSession implements QueueSession, TopicSession {
          if (jbd != null) {
             ClientSession.AddressQuery response = session.addressQuery(jbd.getSimpleAddress());
 
-            if (!response.isExists() && response.isAutoCreateJmsQueues()) {
-               if (jbd.isQueue()) {
-                  session.createAddress(jbd.getSimpleAddress(), false);
-                  session.createQueue(jbd.getSimpleAddress(), jbd.getSimpleAddress(), null,
true);
-               } else {
-                  session.createAddress(jbd.getSimpleAddress(), true);
+            if (jbd.isQueue()) {
+               if (!response.isExists()) {
+                  if (response.isAutoCreateJmsQueues()) {
+                     session.createAddress(jbd.getSimpleAddress(), false);
+                  } else {
+                     throw new InvalidDestinationException("Destination " + jbd.getName()
+ " does not exist");
+                  }
                }
 
-            } else if (!response.isExists() && !response.isAutoCreateJmsQueues())
{
-               throw new InvalidDestinationException("Destination " + jbd.getName() + " does
not exist");
+               if (response.getQueueNames().isEmpty()) {
+                  if (response.isAutoCreateJmsQueues()) {
+                     session.createQueue(jbd.getSimpleAddress(), jbd.getSimpleAddress(),
null, true);
+                  } else {
+                     throw new InvalidDestinationException("Destination " + jbd.getName()
+ " does not exist");
+                  }
+               }
+            } else {
+               if (!response.isExists()) {
+                  if (response.isAutoCreateJmsTopics()) {
+                     session.createAddress(jbd.getSimpleAddress(), true);
+                  } else {
+                     throw new InvalidDestinationException("Destination " + jbd.getName()
+ " does not exist");
+                  }
+               }
             }
-
-            connection.addKnownDestination(jbd.getSimpleAddress());
          }
 
          ClientProducer producer = session.createProducer(jbd == null ? null : jbd.getSimpleAddress());

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java
index b8cd3b0..d1a1871 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java
@@ -1038,6 +1038,8 @@ public interface Configuration {
 
    Configuration setMaxDiskUsage(int maxDiskUsage);
 
+   ConfigurationImpl setInternalNamingPrefix(String internalNamingPrefix);
+
    Configuration setDiskScanPeriod(int diskScanPeriod);
 
    int getDiskScanPeriod();
@@ -1078,4 +1080,5 @@ public interface Configuration {
 
    Configuration setNetworkCheckPing6Command(String command);
 
+   String getInternalNamingPrefix();
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java
index 604526c..f266eb4 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java
@@ -279,6 +279,8 @@ public class ConfigurationImpl implements Configuration, Serializable
{
 
    private String networkCheckPing6Command = NetworkHealthCheck.IPV6_DEFAULT_COMMAND;
 
+   private String internalNamingPrefix = ActiveMQDefaultConfiguration.getInternalNamingPrefix();
+
    /**
     * Parent folder for all data folders.
     */
@@ -1901,6 +1903,17 @@ public class ConfigurationImpl implements Configuration, Serializable
{
    }
 
    @Override
+   public String getInternalNamingPrefix() {
+      return internalNamingPrefix;
+   }
+
+   @Override
+   public ConfigurationImpl setInternalNamingPrefix(String internalNamingPrefix) {
+      this.internalNamingPrefix = internalNamingPrefix;
+      return this;
+   }
+
+   @Override
    public ConfigurationImpl setDiskScanPeriod(int diskScanPeriod) {
       this.diskScanPeriod = diskScanPeriod;
       return this;

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
index f06982b..cebf260 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
@@ -183,6 +183,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil
{
 
    private static final String DISK_SCAN_PERIOD = "disk-scan-period";
 
+   private static final String INTERNAL_NAMING_PREFIX = "internal-naming-prefix";
+
    // Attributes ----------------------------------------------------
 
    private boolean validateAIO = false;
@@ -298,6 +300,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil
{
 
       config.setDiskScanPeriod(getInteger(e, DISK_SCAN_PERIOD, config.getDiskScanPeriod(),
Validators.MINUS_ONE_OR_GT_ZERO));
 
+      config.setInternalNamingPrefix(getString(e, INTERNAL_NAMING_PREFIX, config.getInternalNamingPrefix(),
Validators.NO_CHECK));
+
       // parsing cluster password
       String passwordText = getString(e, "cluster-password", null, Validators.NO_CHECK);
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
index 4c5b566..369fbc9 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
@@ -473,7 +473,11 @@ public interface ActiveMQServer extends ActiveMQComponent {
 
    void removeClientConnection(String clientId);
 
+   AddressInfo putAddressInfoIfAbsent(AddressInfo addressInfo) throws Exception;
+
    AddressInfo createOrUpdateAddressInfo(AddressInfo addressInfo) throws Exception;
 
    AddressInfo removeAddressInfo(SimpleString address) throws Exception;
+
+   String getInternalNamingPrefix();
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java
index 3b35c14..969ab42 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java
@@ -78,6 +78,8 @@ public class ClusterConnectionBridge extends BridgeImpl {
 
    private final ServerLocatorInternal discoveryLocator;
 
+   private final String storeAndForwardPrefix;
+
    public ClusterConnectionBridge(final ClusterConnection clusterConnection,
                                   final ClusterManager clusterManager,
                                   final ServerLocatorInternal targetLocator,
@@ -104,7 +106,8 @@ public class ClusterConnectionBridge extends BridgeImpl {
                                   final SimpleString managementAddress,
                                   final SimpleString managementNotificationAddress,
                                   final MessageFlowRecord flowRecord,
-                                  final TransportConfiguration connector) {
+                                  final TransportConfiguration connector,
+                                  final String storeAndForwardPrefix) {
       super(targetLocator, initialConnectAttempts, reconnectAttempts, 0, // reconnectAttemptsOnSameNode
means nothing on the clustering bridge since we always try the same
             retryInterval, retryMultiplier, maxRetryInterval, nodeUUID, name, queue, executor,
filterString, forwardingAddress, scheduledExecutor, transformer, useDuplicateDetection, user,
password, storageManager);
 
@@ -128,6 +131,8 @@ public class ClusterConnectionBridge extends BridgeImpl {
       if (logger.isTraceEnabled()) {
          logger.trace("Setting up bridge between " + clusterConnection.getConnector() + "
and " + targetLocator, new Exception("trace"));
       }
+
+      this.storeAndForwardPrefix = storeAndForwardPrefix;
    }
 
    @Override
@@ -216,6 +221,8 @@ public class ClusterConnectionBridge extends BridgeImpl {
 
          SimpleString notifQueueName = new SimpleString(qName);
 
+         String filterString = flowRecord.getAddress();
+
          SimpleString filter = new SimpleString(ManagementHelper.HDR_BINDING_TYPE + "<>"
+
                                                    BindingType.DIVERT.toInt() +
                                                    " AND " +
@@ -239,7 +246,7 @@ public class ClusterConnectionBridge extends BridgeImpl {
                                                    "<" +
                                                    flowRecord.getMaxHops() +
                                                    " AND (" +
-                                                   createSelectorFromAddress(flowRecord.getAddress())
+
+                                                   createSelectorFromAddress(appendIgnoresToFilter(flowRecord.getAddress()))
+
                                                    ")");
 
          sessionConsumer.createTemporaryQueue(managementNotificationAddress, notifQueueName,
filter);
@@ -266,6 +273,7 @@ public class ClusterConnectionBridge extends BridgeImpl {
       }
    }
 
+
    /**
     * Takes in a string of an address filter or comma separated list and generates an appropriate
JMS selector for
     * filtering queues.
@@ -332,6 +340,16 @@ public class ClusterConnectionBridge extends BridgeImpl {
       return builder.toString();
    }
 
+   private String appendIgnoresToFilter(String filterString) {
+      if (filterString != null && !filterString.isEmpty()) {
+         filterString += ",";
+      }
+      filterString += "!" + storeAndForwardPrefix;
+      filterString += ",!" + managementAddress;
+      filterString += ",!" + managementNotificationAddress;
+      return filterString;
+   }
+
    @Override
    protected void afterConnect() throws Exception {
       super.afterConnect();

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionImpl.java
index 4b9f0b7..01f7b28 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionImpl.java
@@ -76,6 +76,7 @@ public final class ClusterConnectionImpl implements ClusterConnection, AfterConn
 
    private static final Logger logger = Logger.getLogger(ClusterConnectionImpl.class);
 
+   private static final String SN_PREFIX = "sf.";
    /**
     * When getting member on node-up and down we have to remove the name from the transport
config
     * as the setting we build here doesn't need to consider the name, so use the same name
on all
@@ -170,6 +171,8 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
 
    private final int clusterNotificationAttempts;
 
+   private final String storeAndForwardPrefix;
+
    public ClusterConnectionImpl(final ClusterManager manager,
                                 final TransportConfiguration[] staticTranspConfigs,
                                 final TransportConfiguration connector,
@@ -277,6 +280,7 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
          }
       }
 
+      this.storeAndForwardPrefix = server.getInternalNamingPrefix() + SN_PREFIX;
    }
 
    public ClusterConnectionImpl(final ClusterManager manager,
@@ -375,6 +379,8 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
       clusterConnector = new DiscoveryClusterConnector(dg);
 
       this.manager = manager;
+
+      this.storeAndForwardPrefix = server.getInternalNamingPrefix() + SN_PREFIX;
    }
 
    @Override
@@ -702,7 +708,7 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
 
                // New node - create a new flow record
 
-               final SimpleString queueName = new SimpleString("sf." + name + "." + nodeID);
+               final SimpleString queueName = new SimpleString(storeAndForwardPrefix + name
+ "." + nodeID);
 
                Binding queueBinding = postOffice.getBinding(queueName);
 
@@ -799,7 +805,7 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
       targetLocator.addIncomingInterceptor(new IncomingInterceptorLookingForExceptionMessage(manager,
executorFactory.getExecutor()));
       MessageFlowRecordImpl record = new MessageFlowRecordImpl(targetLocator, eventUID, targetNodeID,
connector, queueName, queue);
 
-      ClusterConnectionBridge bridge = new ClusterConnectionBridge(this, manager, targetLocator,
serverLocator, initialConnectAttempts, reconnectAttempts, retryInterval, retryIntervalMultiplier,
maxRetryInterval, nodeManager.getUUID(), record.getEventUID(), record.getTargetNodeID(), record.getQueueName(),
record.getQueue(), executorFactory.getExecutor(), null, null, scheduledExecutor, null, useDuplicateDetection,
clusterUser, clusterPassword, server.getStorageManager(), managementService.getManagementAddress(),
managementService.getManagementNotificationAddress(), record, record.getConnector());
+      ClusterConnectionBridge bridge = new ClusterConnectionBridge(this, manager, targetLocator,
serverLocator, initialConnectAttempts, reconnectAttempts, retryInterval, retryIntervalMultiplier,
maxRetryInterval, nodeManager.getUUID(), record.getEventUID(), record.getTargetNodeID(), record.getQueueName(),
record.getQueue(), executorFactory.getExecutor(), null, null, scheduledExecutor, null, useDuplicateDetection,
clusterUser, clusterPassword, server.getStorageManager(), managementService.getManagementAddress(),
managementService.getManagementNotificationAddress(), record, record.getConnector(), storeAndForwardPrefix);
 
       targetLocator.setIdentity("(Cluster-connection-bridge::" + bridge.toString() + "::"
+ this.toString() + ")");
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
index 67e60f1..b09f04d 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
@@ -172,6 +172,8 @@ public class ActiveMQServerImpl implements ActiveMQServer {
 
    private static final Logger logger = Logger.getLogger(ActiveMQServerImpl.class);
 
+   public static final String INTERNAL_NAMING_PREFIX = "$.artemis.internal";
+
    /**
     * JMS Topics (which are outside of the scope of the core API) will require a dumb subscription
     * with a dummy-filter at this current version as a way to keep its existence valid and
TCK
@@ -719,7 +721,11 @@ public class ActiveMQServerImpl implements ActiveMQServer {
          }
       }
 
-      return new BindingQueryResult(!names.isEmpty(), names, autoCreateJmsQueues, autoCreateJmsTopics);
+      if (autoCreateJmsTopics) {
+         putAddressInfoIfAbsent(new AddressInfo(address));
+      }
+
+      return new BindingQueryResult(getAddressInfo(address) != null, names, autoCreateJmsQueues,
autoCreateJmsTopics);
    }
 
    @Override
@@ -2220,14 +2226,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
 
    private void deployQueuesFromListCoreQueueConfiguration(List<CoreQueueConfiguration>
queues) throws Exception {
       for (CoreQueueConfiguration config : queues) {
-         deployQueue(SimpleString.toSimpleString(config.getAddress()),
-                     SimpleString.toSimpleString(config.getName()),
-                     SimpleString.toSimpleString(config.getFilterString()),
-                     config.isDurable(),
-                     false,
-                     false,
-                     config.getMaxConsumers(),
-                     config.getDeleteOnNoConsumers());
+         deployQueue(SimpleString.toSimpleString(config.getAddress()), SimpleString.toSimpleString(config.getName()),
SimpleString.toSimpleString(config.getFilterString()), config.isDurable(), false, false, config.getMaxConsumers(),
config.getDeleteOnNoConsumers());
       }
    }
 
@@ -2331,6 +2330,18 @@ public class ActiveMQServerImpl implements ActiveMQServer {
    }
 
    @Override
+   public AddressInfo putAddressInfoIfAbsent(AddressInfo addressInfo) throws Exception {
+      AddressInfo result = postOffice.addAddressInfo(addressInfo);
+
+      // TODO: is this the right way to do this?
+      long txID = storageManager.generateID();
+      storageManager.addAddressBinding(txID, addressInfo);
+      storageManager.commitBindings(txID);
+
+      return result;
+   }
+
+   @Override
    public AddressInfo createOrUpdateAddressInfo(AddressInfo addressInfo) throws Exception
{
       AddressInfo result = postOffice.addOrUpdateAddressInfo(addressInfo);
 
@@ -2355,6 +2366,11 @@ public class ActiveMQServerImpl implements ActiveMQServer {
    }
 
    @Override
+   public String getInternalNamingPrefix() {
+      return configuration.getInternalNamingPrefix();
+   }
+
+   @Override
    public AddressInfo getAddressInfo(SimpleString address) {
       return postOffice.getAddressInfo(address);
    }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java
index 9c1e209..2c28676 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java
@@ -128,7 +128,8 @@ public class ScaleDownHandler {
                }
             }
 
-            if (address.toString().startsWith("sf.")) {
+            String sfPrefix =  ((PostOfficeImpl) postOffice).getServer().getInternalNamingPrefix()
+ "sf.";
+            if (address.toString().startsWith(sfPrefix)) {
                messageCount += scaleDownSNF(address, queues, producer);
             } else {
                messageCount += scaleDownRegularMessages(address, queues, session, producer);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-server/src/main/resources/schema/artemis-configuration.xsd
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
index 4902bd3..232a81b 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -42,6 +42,16 @@
             </xsd:annotation>
          </xsd:element>
 
+         <xsd:element name="internal-naming-prefix" type="xsd:string" maxOccurs="1" minOccurs="0">
+            <xsd:annotation>
+               <xsd:documentation>
+                  Artemis uses internal queues and addresses for implmenting certain behaviours.
 These queues and addresses
+                  will be prefixed by default with "$.activemq.internal" to avoid naming
clashes with user namespacing.
+                  This can be overriden by setting this value to a valid Artemis address.
+               </xsd:documentation>
+            </xsd:annotation>
+         </xsd:element>
+
          <xsd:element name="resolve-protocols" type="xsd:boolean" default="true" maxOccurs="1"
                       minOccurs="0">
             <xsd:annotation>
@@ -871,66 +881,6 @@
             </xsd:complexType>
          </xsd:element>
 
-         <xsd:element name="network-check-list" type="xsd:string" default="" maxOccurs="1"
minOccurs="0">
-            <xsd:annotation>
-               <xsd:documentation>
-                  A comma separated list of IPs to be used to validate if the broker should
be kept up
-               </xsd:documentation>
-            </xsd:annotation>
-         </xsd:element>
-
-         <xsd:element name="network-check-URL-list" type="xsd:string" default="" maxOccurs="1"
minOccurs="0">
-            <xsd:annotation>
-               <xsd:documentation>
-                  A comma separated list of URLs to be used to validate if the broker should
be kept up
-               </xsd:documentation>
-            </xsd:annotation>
-         </xsd:element>
-
-         <xsd:element name="network-check-period" type="xsd:long" default="10000" maxOccurs="1"
minOccurs="0">
-            <xsd:annotation>
-               <xsd:documentation>
-                  A frequency in milliseconds to how often we should check if the network
is still up
-               </xsd:documentation>
-            </xsd:annotation>
-         </xsd:element>
-
-         <xsd:element name="network-check-timeout" type="xsd:long" default="1000" maxOccurs="1"
minOccurs="0">
-            <xsd:annotation>
-               <xsd:documentation>
-                  A timeout used in milliseconds to be used on the ping.
-               </xsd:documentation>
-            </xsd:annotation>
-         </xsd:element>
-
-         <xsd:element name="network-check-NIC" type="xsd:string" default="" maxOccurs="1"
minOccurs="0">
-            <xsd:annotation>
-               <xsd:documentation>
-                  The network interface card name to be used to validate the address.
-               </xsd:documentation>
-            </xsd:annotation>
-         </xsd:element>
-
-
-         <xsd:element name="network-check-ping-command" type="xsd:string" default="" maxOccurs="1"
minOccurs="0">
-            <xsd:annotation>
-               <xsd:documentation>
-                  The ping command used to ping IPV4 addresses.
-               </xsd:documentation>
-            </xsd:annotation>
-         </xsd:element>
-
-
-         <xsd:element name="network-check-ping6-command" type="xsd:string" default=""
maxOccurs="1" minOccurs="0">
-            <xsd:annotation>
-               <xsd:documentation>
-                  The ping command used to ping IPV6 addresses.
-               </xsd:documentation>
-            </xsd:annotation>
-         </xsd:element>
-
-
-
          <xsd:element name="addresses" type="addressesType" maxOccurs="1" minOccurs="0"
/>
       </xsd:all>
    </xsd:complexType>

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/artemis-tools/src/test/resources/artemis-configuration.xsd
----------------------------------------------------------------------
diff --git a/artemis-tools/src/test/resources/artemis-configuration.xsd b/artemis-tools/src/test/resources/artemis-configuration.xsd
index 4c3e068..e8b9161 100644
--- a/artemis-tools/src/test/resources/artemis-configuration.xsd
+++ b/artemis-tools/src/test/resources/artemis-configuration.xsd
@@ -1265,7 +1265,7 @@
 
    <xsd:complexType name="cluster-connectionType">
       <xsd:sequence>
-         <xsd:element name="address" type="xsd:string" maxOccurs="1" minOccurs="1">
+         <xsd:element name="address" type="xsd:string" maxOccurs="1" minOccurs="0">
             <xsd:annotation>
                <xsd:documentation>
                   name of the address this cluster connection applies to

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/AMQPToOpenwireTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/AMQPToOpenwireTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/AMQPToOpenwireTest.java
index 0374aef..0bc3e28 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/AMQPToOpenwireTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/AMQPToOpenwireTest.java
@@ -26,6 +26,7 @@ import javax.jms.Session;
 import java.io.ByteArrayInputStream;
 import java.io.ObjectInputStream;
 import java.util.ArrayList;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.ActiveMQXAConnectionFactory;
@@ -60,23 +61,21 @@ public class AMQPToOpenwireTest extends ActiveMQTestBase {
    public void setUp() throws Exception {
       super.setUp();
       server = createServer(true, true);
-      serverManager = new JMSServerManagerImpl(server);
+      server.start();
+      server.waitForActivation(10, TimeUnit.SECONDS);
+
       Configuration serverConfig = server.getConfiguration();
       serverConfig.getAddressesSettings().put("#", new AddressSettings().setAutoCreateJmsQueues(false).setDeadLetterAddress(new
SimpleString("ActiveMQ.DLQ")));
       serverConfig.setSecurityEnabled(false);
-      serverManager.start();
       coreQueue = new SimpleString(queueName);
-      this.server.createQueue(coreQueue, coreQueue, null, false, false);
+      server.createQueue(coreQueue, coreQueue, null, false, false);
       qpidfactory = new JmsConnectionFactory("amqp://localhost:61616");
    }
 
    @Override
    @After
    public void tearDown() throws Exception {
-      if (serverManager != null) {
-         serverManager.stop();
-         serverManager = null;
-      }
+      server.stop();
    }
 
    @Test

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTest.java
index 486c2d5..00d220b 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTest.java
@@ -1042,7 +1042,7 @@ public class MQTTTest extends MQTTTestSupport {
       initializeConnection(provider);
 
       // send retained message
-      final String address = "jms/queue/" + mqttAddress;
+      final String address = mqttAddress;
       final String RETAINED = "RETAINED";
 
       final byte[] payload = RETAINED.getBytes();
@@ -1088,7 +1088,7 @@ public class MQTTTest extends MQTTTestSupport {
    public void doTestSendJMSReceiveMQTT(String destinationName) throws Exception {
       final MQTTClientProvider provider = getMQTTClientProvider();
       initializeConnection(provider);
-      provider.subscribe("jms/queue/foo/+", AT_MOST_ONCE);
+      provider.subscribe("foo/+", AT_MOST_ONCE);
 
       Connection connection = cf.createConnection();
       connection.start();
@@ -1279,7 +1279,7 @@ public class MQTTTest extends MQTTTestSupport {
 
       // publish
       for (int i = 0; i < messagesToSend; ++i) {
-         connection.publish("jms/queue/test/foo", "hello world".getBytes(), QoS.AT_LEAST_ONCE,
false);
+         connection.publish("test/foo", "hello world".getBytes(), QoS.AT_LEAST_ONCE, false);
       }
 
       connection.disconnect();

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDown3NodeTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDown3NodeTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDown3NodeTest.java
index be91bdd..d13e876 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDown3NodeTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDown3NodeTest.java
@@ -117,7 +117,7 @@ public class ScaleDown3NodeTest extends ClusterTestBase {
       createQueue(2, addressName, queueName1, null, false, servers[2].getConfiguration().getClusterUser(),
servers[2].getConfiguration().getClusterPassword());
 
       // pause the SnF queue so that when the server tries to redistribute a message it won't
actually go across the cluster bridge
-      String snfAddress = "sf.cluster0." + servers[0].getNodeID().toString();
+      String snfAddress = servers[0].getInternalNamingPrefix() + "sf.cluster0." + servers[0].getNodeID().toString();
       Queue snfQueue = ((LocalQueueBinding) servers[2].getPostOffice().getBinding(SimpleString.toSimpleString(snfAddress))).getQueue();
       snfQueue.pause();
 
@@ -260,7 +260,7 @@ public class ScaleDown3NodeTest extends ClusterTestBase {
       createQueue(2, addressName, queueName3, null, false, servers[2].getConfiguration().getClusterUser(),
servers[2].getConfiguration().getClusterPassword());
 
       // pause the SnF queue so that when the server tries to redistribute a message it won't
actually go across the cluster bridge
-      String snfAddress = "sf.cluster0." + servers[0].getNodeID().toString();
+      String snfAddress = servers[0].getInternalNamingPrefix() + "sf.cluster0." + servers[0].getNodeID().toString();
       Queue snfQueue = ((LocalQueueBinding) servers[2].getPostOffice().getBinding(SimpleString.toSimpleString(snfAddress))).getQueue();
       snfQueue.pause();
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/0006627d/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDownTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDownTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDownTest.java
index d2c6fc2..cba3008 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDownTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDownTest.java
@@ -163,7 +163,7 @@ public class ScaleDownTest extends ClusterTestBase {
       for (Map.Entry<SimpleString, Binding> entry : servers[0].getPostOffice().getAllBindings().entrySet())
{
          String temp = entry.getValue().getAddress().toString();
 
-         if (temp.startsWith("sf.") && temp.endsWith(servers[1].getNodeID().toString()))
{
+         if (temp.startsWith(servers[1].getInternalNamingPrefix() + "sf.") && temp.endsWith(servers[1].getNodeID().toString()))
{
             // we found the sf queue for the other node
             // need to pause the sfQueue here
             ((LocalQueueBinding) entry.getValue()).getQueue().pause();


Mime
View raw message