You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ma...@apache.org on 2016/11/07 12:23:48 UTC
[1/2] activemq-artemis git commit: fix some tests
Repository: activemq-artemis
Updated Branches:
refs/heads/ARTEMIS-780 70aa8bc6c -> dc9c07878
fix some tests
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/dc9c0787
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/dc9c0787
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/dc9c0787
Branch: refs/heads/ARTEMIS-780
Commit: dc9c078780ef88b7a5c6dc92c7ab5d85b913c517
Parents: 362b2f2
Author: Martyn Taylor <mt...@redhat.com>
Authored: Mon Nov 7 12:19:52 2016 +0000
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Mon Nov 7 12:20:07 2016 +0000
----------------------------------------------------------------------
.../integration/crossprotocol/AMQPToOpenwireTest.java | 13 ++++++-------
.../tests/integration/mqtt/imported/MQTTTest.java | 6 +++---
2 files changed, 9 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/dc9c0787/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/dc9c0787/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();
[2/2] activemq-artemis git commit: Fix Clustering Addressing
Posted by ma...@apache.org.
Fix Clustering Addressing
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/362b2f23
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/362b2f23
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/362b2f23
Branch: refs/heads/ARTEMIS-780
Commit: 362b2f23d271ce2862893b3ed69abcd20a60f573
Parents: 70aa8bc
Author: Martyn Taylor <mt...@redhat.com>
Authored: Mon Nov 7 11:36:18 2016 +0000
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Mon Nov 7 12:20:07 2016 +0000
----------------------------------------------------------------------
.../config/ActiveMQDefaultConfiguration.java | 6 ++++++
.../artemis/core/config/Configuration.java | 3 +++
.../core/config/impl/ConfigurationImpl.java | 13 ++++++++++++
.../deployers/impl/FileConfigurationParser.java | 4 ++++
.../artemis/core/server/ActiveMQServer.java | 2 ++
.../cluster/impl/ClusterConnectionBridge.java | 22 ++++++++++++++++++--
.../cluster/impl/ClusterConnectionImpl.java | 8 ++++++-
.../core/server/impl/ActiveMQServerImpl.java | 7 +++++++
.../resources/schema/artemis-configuration.xsd | 10 +++++++++
9 files changed, 72 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/362b2f23/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 f9861a4..5511ab6 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
@@ -441,6 +441,8 @@ public final class ActiveMQDefaultConfiguration {
public static final boolean DEFAULT_DELETE_QUEUE_ON_NO_CONSUMERS = false;
+ 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.
*/
@@ -1186,4 +1188,8 @@ public final class ActiveMQDefaultConfiguration {
public static boolean getDefaultDeleteQueueOnNoConsumers() {
return DEFAULT_DELETE_QUEUE_ON_NO_CONSUMERS;
}
+
+ public static String getInternalNamingPrefix() {
+ return DEFAULT_INTERNAL_NAMING_PREFIX;
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/362b2f23/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 8d47f97..17e9f6f 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
@@ -1013,8 +1013,11 @@ public interface Configuration {
Configuration setMaxDiskUsage(int maxDiskUsage);
+ ConfigurationImpl setInternalNamingPrefix(String internalNamingPrefix);
+
Configuration setDiskScanPeriod(int diskScanPeriod);
int getDiskScanPeriod();
+ String getInternalNamingPrefix();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/362b2f23/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 3b66f83..37fa687 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
@@ -257,6 +257,8 @@ public class ConfigurationImpl implements Configuration, Serializable {
private int diskScanPeriod = ActiveMQDefaultConfiguration.getDefaultDiskScanPeriod();
+ private String internalNamingPrefix = ActiveMQDefaultConfiguration.getInternalNamingPrefix();
+
/**
* Parent folder for all data folders.
*/
@@ -1826,6 +1828,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/362b2f23/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 a77b850..ac0a7a1 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;
@@ -296,6 +298,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/362b2f23/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 51e1830..bb819ae 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
@@ -468,4 +468,6 @@ public interface ActiveMQServer extends ActiveMQComponent {
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/362b2f23/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/362b2f23/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..aa9f0ce 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
@@ -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/362b2f23/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 2d94fa3..d62598e 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
@@ -170,6 +170,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
@@ -2286,6 +2288,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/362b2f23/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 8da84fe..f288ea6 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -34,6 +34,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>