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/10/24 13:35:07 UTC
[01/17] activemq-artemis git commit: ARTEMIS-812 The countDelta
attribute showing negative values [Forced Update!]
Repository: activemq-artemis
Updated Branches:
refs/heads/ARTEMIS-780 60bbb3fbd -> 6d41d37e9 (forced update)
ARTEMIS-812 The countDelta attribute showing negative values
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/9c16ba26
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/9c16ba26
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/9c16ba26
Branch: refs/heads/ARTEMIS-780
Commit: 9c16ba26becebd69547e6eb16d384cdd58d2cc45
Parents: e44c99d
Author: Howard Gao <ho...@gmail.com>
Authored: Fri Oct 21 17:13:41 2016 +0800
Committer: Howard Gao <ho...@gmail.com>
Committed: Fri Oct 21 17:13:41 2016 +0800
----------------------------------------------------------------------
.../impl/ActiveMQServerControlImpl.java | 5 +-
.../paging/cursor/PageSubscriptionCounter.java | 3 +
.../impl/PageSubscriptionCounterImpl.java | 14 +-
.../core/paging/impl/PagingStoreImpl.java | 1 -
.../core/server/ActiveMQMessageBundle.java | 4 +-
.../core/server/ActiveMQServerLogger.java | 5 +
.../artemis/core/server/impl/QueueImpl.java | 15 +-
.../ClusteredMessageCounterTest.java | 318 +++++++++++++++++++
.../management/ActiveMQServerControlTest.java | 9 +-
9 files changed, 359 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9c16ba26/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java
index b96fcbf..5918ec4 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java
@@ -867,7 +867,10 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
clearIO();
try {
if (newPeriod < MessageCounterManagerImpl.MIN_SAMPLE_PERIOD) {
- throw ActiveMQMessageBundle.BUNDLE.invalidMessageCounterPeriod(MessageCounterManagerImpl.MIN_SAMPLE_PERIOD);
+ if (newPeriod <= 0) {
+ throw ActiveMQMessageBundle.BUNDLE.periodMustGreaterThanZero(newPeriod);
+ }
+ ActiveMQServerLogger.LOGGER.invalidMessageCounterPeriod(newPeriod);
}
if (messageCounterManager != null && newPeriod != messageCounterManager.getSamplePeriod()) {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9c16ba26/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscriptionCounter.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscriptionCounter.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscriptionCounter.java
index 343f936..2e1d7b6 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscriptionCounter.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscriptionCounter.java
@@ -21,6 +21,9 @@ import org.apache.activemq.artemis.core.transaction.Transaction;
public interface PageSubscriptionCounter {
+ //incremental counter of messages added
+ long getValueAdded();
+
long getValue();
void increment(Transaction tx, int add) throws Exception;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9c16ba26/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionCounterImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionCounterImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionCounterImpl.java
index 92f313b..e01098d 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionCounterImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionCounterImpl.java
@@ -61,6 +61,8 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
private final AtomicLong value = new AtomicLong(0);
+ private final AtomicLong added = new AtomicLong(0);
+
private final AtomicLong pendingValue = new AtomicLong(0);
private final LinkedList<Long> incrementRecords = new LinkedList<>();
@@ -93,6 +95,11 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
}
@Override
+ public long getValueAdded() {
+ return added.get() + pendingValue.get();
+ }
+
+ @Override
public long getValue() {
return value.get() + pendingValue.get();
}
@@ -205,6 +212,7 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
this.subscription.notEmpty();
}
this.value.set(value1);
+ this.added.set(value1);
this.recordID = recordID1;
}
@@ -243,6 +251,7 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
recordID = -1;
value.set(0);
+ added.set(0);
incrementRecords.clear();
}
} finally {
@@ -269,6 +278,7 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
for (Pair<Long, Integer> incElement : loadList) {
value.addAndGet(incElement.getB());
+ added.addAndGet(incElement.getB());
incrementRecords.add(incElement.getA());
}
loadList.clear();
@@ -279,7 +289,9 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
@Override
public synchronized void addInc(long id, int variance) {
value.addAndGet(variance);
-
+ if (variance > 0) {
+ added.addAndGet(variance);
+ }
if (id >= 0) {
incrementRecords.add(id);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9c16ba26/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
index f756edd..f05ace0 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
@@ -876,7 +876,6 @@ public class PagingStoreImpl implements PagingStore {
}
for (org.apache.activemq.artemis.core.server.Queue q : nonDurableQueues) {
- q.getPageSubscription().getCounter().increment(tx, 1);
q.getPageSubscription().notEmpty();
ids[i++] = q.getID();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9c16ba26/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java
index c87bd11..f22873b 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java
@@ -194,8 +194,8 @@ public interface ActiveMQMessageBundle {
@Message(id = 119046, value = "invalid value: {0} count must be greater than 0", format = Message.Format.MESSAGE_FORMAT)
IllegalArgumentException greaterThanZero(Integer count);
- @Message(id = 119047, value = "Cannot set Message Counter Sample Period < {0}ms", format = Message.Format.MESSAGE_FORMAT)
- IllegalArgumentException invalidMessageCounterPeriod(Long period);
+ @Message(id = 119047, value = "invalid value: {0} sample period must be greater than 0", format = Message.Format.MESSAGE_FORMAT)
+ IllegalArgumentException periodMustGreaterThanZero(Long newPeriod);
@Message(id = 119048, value = "invalid new Priority value: {0}. It must be between 0 and 9 (both included)", format = Message.Format.MESSAGE_FORMAT)
IllegalArgumentException invalidNewPriority(Integer period);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9c16ba26/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
index ae07a8f..24432a3 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
@@ -1516,4 +1516,9 @@ public interface ActiveMQServerLogger extends BasicLogger {
@LogMessage(level = Logger.Level.ERROR)
@Message(id = 224069, value = "Change detected in broker configuration file, but reload failed", format = Message.Format.MESSAGE_FORMAT)
void configurationReloadFailed(@Cause Throwable t);
+
+ @LogMessage(level = Logger.Level.WARN)
+ @Message(id = 224072, value = "Message Counter Sample Period too short: {0}", format = Message.Format.MESSAGE_FORMAT)
+ void invalidMessageCounterPeriod(long value);
+
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9c16ba26/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index d515b3d..7c8ad0a 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -518,7 +518,9 @@ public class QueueImpl implements Queue {
directDeliver = false;
- messagesAdded++;
+ if (!ref.isPaged()) {
+ messagesAdded++;
+ }
}
@Override
@@ -573,7 +575,9 @@ public class QueueImpl implements Queue {
protected boolean scheduleIfPossible(MessageReference ref) {
if (scheduledDeliveryHandler.checkAndSchedule(ref, true)) {
synchronized (this) {
- messagesAdded++;
+ if (!ref.isPaged()) {
+ messagesAdded++;
+ }
}
return true;
@@ -1165,7 +1169,7 @@ public class QueueImpl implements Queue {
@Override
public long getMessagesAdded() {
if (pageSubscription != null) {
- return messagesAdded + pageSubscription.getCounter().getValue() - pagedReferences.get();
+ return messagesAdded + pageSubscription.getCounter().getValueAdded();
} else {
return messagesAdded;
}
@@ -1819,7 +1823,10 @@ public class QueueImpl implements Queue {
while ((ref = intermediateMessageReferences.poll()) != null) {
internalAddTail(ref);
- messagesAdded++;
+ if (!ref.isPaged()) {
+ messagesAdded++;
+ }
+
if (added++ > MAX_DELIVERIES_IN_LOOP) {
// if we just keep polling from the intermediate we could starve in case there's a sustained load
deliverAsync();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9c16ba26/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/ClusteredMessageCounterTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/ClusteredMessageCounterTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/ClusteredMessageCounterTest.java
new file mode 100644
index 0000000..a8ad897
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/ClusteredMessageCounterTest.java
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.cluster.distribution;
+
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.client.ClientConsumer;
+import org.apache.activemq.artemis.api.core.client.ClientMessage;
+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.management.ActiveMQServerControl;
+import org.apache.activemq.artemis.api.core.management.MessageCounterInfo;
+import org.apache.activemq.artemis.api.core.management.QueueControl;
+import org.apache.activemq.artemis.api.core.management.ResourceNames;
+import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
+import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class ClusteredMessageCounterTest extends ClusterTestBase {
+ private AtomicInteger total = new AtomicInteger();
+ private AtomicBoolean stopFlag = new AtomicBoolean();
+ private Timer timer1 = new Timer();
+ private Timer timer2 = new Timer();
+ private int numMsg = 1000;
+ private List<MessageCounterInfo> results = new ArrayList<>();
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+
+ setupServers();
+ setupClusters();
+ total.set(0);
+ stopFlag.set(false);
+ }
+
+ @Override
+ @After
+ public void tearDown() throws Exception {
+ timer1.cancel();
+ timer2.cancel();
+ super.tearDown();
+ }
+
+ protected void setupServers() throws Exception {
+ setupServer(0, isFileStorage(), isNetty());
+ setupServer(1, isFileStorage(), isNetty());
+ }
+
+ protected void setupClusters() {
+ setupClusterConnection("cluster0", 0, 1, "queues", MessageLoadBalancingType.ON_DEMAND, 1, isNetty(), false);
+ setupClusterConnection("cluster1", 1, 0, "queues", MessageLoadBalancingType.ON_DEMAND, 1, isNetty(), false);
+ }
+
+ protected boolean isNetty() {
+ return true;
+ }
+
+ @Override
+ protected ConfigurationImpl createBasicConfig(final int serverID) {
+ ConfigurationImpl config = super.createBasicConfig(serverID);
+ Map<String, AddressSettings> addrSettingsMap = config.getAddressesSettings();
+ AddressSettings addrSettings = new AddressSettings();
+ addrSettings.setMaxSizeBytes(10 * 1024);
+ addrSettings.setPageSizeBytes(5 * 1024);
+ addrSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addrSettingsMap.put("queues", addrSettings);
+ if (serverID == 1) {
+ config.setMessageCounterEnabled(true);
+ }
+ return config;
+ }
+
+ @Test
+ public void testNonDurableMessageAddedWithPaging() throws Exception {
+ testMessageAddedWithPaging(false);
+ }
+
+ @Test
+ public void testDurableMessageAddedWithPaging() throws Exception {
+ testMessageAddedWithPaging(true);
+ }
+
+ //messages flow from one node to another, in paging mode
+ //check the messageAdded is correct.
+ private void testMessageAddedWithPaging(boolean durable) throws Exception {
+ startServers(0, 1);
+ numMsg = 100;
+
+ try {
+ setupSessionFactory(0, isNetty());
+ setupSessionFactory(1, isNetty());
+
+ createQueue(0, "queues", "queue0", null, false);
+ createQueue(1, "queues", "queue0", null, false);
+
+ waitForBindings(1, "queues", 1, 0, true);
+ waitForBindings(0, "queues", 1, 0, false);
+
+ addConsumer(1, 1, "queue0", null);
+
+ System.out.println("sending.....");
+ send(0, "queues", numMsg, durable, null);
+
+ verifyReceiveAllOnSingleConsumer(true, numMsg, 1);
+
+ QueueControl control = (QueueControl) servers[1].getManagementService().getResource(ResourceNames.CORE_QUEUE + "queue0");
+
+ //wait up to 30sec to allow the counter get updated
+ long timeout = 30000;
+ while (timeout > 0 && (numMsg != control.getMessagesAdded())) {
+ Thread.sleep(1000);
+ timeout -= 1000;
+ }
+ assertEquals(numMsg, control.getMessagesAdded());
+ } finally {
+ stopServers(0, 1);
+ }
+ }
+
+ @Test
+ public void testMessageCounterWithPaging() throws Exception {
+ startServers(0, 1);
+
+ try {
+ setupSessionFactory(0, isNetty());
+ setupSessionFactory(1, isNetty());
+
+ createQueue(0, "queues", "queue0", null, false);
+ createQueue(1, "queues", "queue0", null, false);
+
+ waitForBindings(1, "queues", 1, 0, true);
+ waitForBindings(0, "queues", 1, 0, false);
+
+ System.out.println("sending.....");
+ Thread sendThread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ send(0, "queues", numMsg, true, null);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ System.out.println("messages sent.");
+ }
+ });
+
+ QueueControl control = (QueueControl) servers[1].getManagementService().getResource(ResourceNames.CORE_QUEUE + "queue0");
+ ActiveMQServerControl serverControl = (ActiveMQServerControl) servers[1].getManagementService().getResource(ResourceNames.CORE_SERVER);
+ serverControl.setMessageCounterSamplePeriod(300);
+
+ CountDownLatch resultLatch = new CountDownLatch(40);
+
+ MessageCounterCollector collector = new MessageCounterCollector(control, resultLatch);
+ timer1.schedule(collector, 0);
+
+ PeriodicalReceiver receiver = new PeriodicalReceiver(50, 1, 100);
+ timer2.schedule(receiver, 0);
+
+ sendThread.start();
+
+ try {
+ resultLatch.await(120, TimeUnit.SECONDS);
+ } finally {
+ stopFlag.set(true);
+ }
+ sendThread.join();
+ System.out.println("Results collected: " + results.size());
+ //checking
+ for (MessageCounterInfo info : results) {
+ assertTrue("countDelta should be positive " + info.getCountDelta() + dumpResults(results), info.getCountDelta() >= 0);
+ }
+ } finally {
+ timer1.cancel();
+ timer2.cancel();
+ stopServers(0, 1);
+ }
+ }
+
+ private String dumpResults(List<MessageCounterInfo> results) {
+ StringBuilder builder = new StringBuilder("\n");
+ for (int i = 0; i < results.size(); i++) {
+ builder.append("result[" + i + "]: " + results.get(i).getCountDelta() + " " + results.get(i).getCount() + "\n");
+ }
+ return builder.toString();
+ }
+
+ //Periodically read the counter
+ private class MessageCounterCollector extends TimerTask {
+ private QueueControl queueControl;
+ private CountDownLatch resultLatch;
+
+ MessageCounterCollector(QueueControl queueControl, CountDownLatch resultLatch) {
+ this.queueControl = queueControl;
+ this.resultLatch = resultLatch;
+ }
+
+ @Override
+ public void run() {
+ if (stopFlag.get()) {
+ return;
+ }
+ try {
+ String result = queueControl.listMessageCounter();
+ MessageCounterInfo info = MessageCounterInfo.fromJSON(result);
+ if (info.getCountDelta() != 0) {
+ System.out.println("non zero value got ---> " + info.getCountDelta());
+ }
+ results.add(info);
+ resultLatch.countDown();
+ if (info.getCountDelta() < 0) {
+ //stop and make the test finish quick
+ stopFlag.set(true);
+ while (resultLatch.getCount() > 0) {
+ resultLatch.countDown();
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ if (!stopFlag.get()) {
+ timer1.schedule(new MessageCounterCollector(this.queueControl, resultLatch), 200);
+ }
+ }
+ }
+ }
+
+ //Peroidically receive a number of messages
+ private class PeriodicalReceiver extends TimerTask {
+ private int batchSize;
+ private int serverID;
+ private long period;
+
+ PeriodicalReceiver(int batchSize, int serverID, long period) {
+ this.batchSize = batchSize;
+ this.serverID = serverID;
+ this.period = period;
+ }
+
+ @Override
+ public void run() {
+ if (stopFlag.get()) {
+ return;
+ }
+ int num = 0;
+ ClientSessionFactory sf = sfs[serverID];
+ ClientSession session = null;
+ ClientConsumer consumer = null;
+ try {
+ session = sf.createSession(false, true, false);
+ consumer = session.createConsumer("queue0", null);
+ session.start();
+ for (; num < batchSize || stopFlag.get(); num++) {
+ ClientMessage message = consumer.receive(2000);
+ if (message == null) {
+ System.out.println("No more messages received!");
+ break;
+ }
+ message.acknowledge();
+ }
+ session.commit();
+ } catch (ActiveMQException e) {
+ e.printStackTrace();
+ } finally {
+ System.out.println("received messages: " + num);
+ if (consumer != null) {
+ try {
+ consumer.close();
+ } catch (ActiveMQException e) {
+ e.printStackTrace();
+ }
+ }
+ if (session != null) {
+ try {
+ session.close();
+ } catch (ActiveMQException e) {
+ e.printStackTrace();
+ }
+ }
+
+ //we only receive (numMsg - 200) to avoid the paging being cleaned up
+ //when all paged messages are consumed.
+ if (!stopFlag.get() && total.addAndGet(num) < numMsg - 200) {
+ System.out.println("go for another batch " + total.get());
+ timer2.schedule(new PeriodicalReceiver(this.batchSize, this.serverID, this.period), period);
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9c16ba26/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlTest.java
index 86d19db..d040b8a 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlTest.java
@@ -414,13 +414,10 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
} catch (Exception e) {
}
- try {
- serverControl.setMessageCounterSamplePeriod(MessageCounterManagerImpl.MIN_SAMPLE_PERIOD - 1);
- Assert.fail();
- } catch (Exception e) {
- }
+ //this only gets warning now and won't cause exception.
+ serverControl.setMessageCounterSamplePeriod(MessageCounterManagerImpl.MIN_SAMPLE_PERIOD - 1);
- Assert.assertEquals(newSample, serverControl.getMessageCounterSamplePeriod());
+ Assert.assertEquals(MessageCounterManagerImpl.MIN_SAMPLE_PERIOD - 1, serverControl.getMessageCounterSamplePeriod());
}
protected void restartServer() throws Exception {
[13/17] activemq-artemis git commit: Fix formatting
Posted by ma...@apache.org.
Fix formatting
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/c2d4f3ee
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/c2d4f3ee
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/c2d4f3ee
Branch: refs/heads/ARTEMIS-780
Commit: c2d4f3ee14741b6cbb06427bff259a2c21287612
Parents: d6e0e6a
Author: jbertram <jb...@apache.com>
Authored: Fri Oct 21 11:25:07 2016 -0500
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Mon Oct 24 14:33:53 2016 +0100
----------------------------------------------------------------------
.../core/persistence/impl/RoutingType.java | 43 --------------------
.../config/XMLConfigurationMigration.java | 26 ++++--------
2 files changed, 8 insertions(+), 61 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/c2d4f3ee/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/RoutingType.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/RoutingType.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/RoutingType.java
deleted file mode 100644
index 329d8e9..0000000
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/RoutingType.java
+++ /dev/null
@@ -1,43 +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.core.persistence.impl;
-
-public enum RoutingType {
- Multicast, Anycast;
-
- public byte getType() {
- switch (this) {
- case Multicast:
- return 0;
- case Anycast:
- return 1;
- default:
- return -1;
- }
- }
-
- public static RoutingType getType(byte type) {
- switch (type) {
- case 0:
- return Multicast;
- case 1:
- return Anycast;
- default:
- return null;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/c2d4f3ee/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigration.java
----------------------------------------------------------------------
diff --git a/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigration.java b/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigration.java
index 56833ea..90be53c 100644
--- a/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigration.java
+++ b/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigration.java
@@ -50,19 +50,16 @@ public class XMLConfigurationMigration {
if (args.length == 0) {
System.err.println("Invalid args");
printUsage();
- }
- else {
+ } else {
File input = new File(args[0]);
if (input.isDirectory()) {
System.out.println("Scanning directory: " + input.getAbsolutePath());
recursiveTransform(input);
- }
- else {
+ } else {
if (args.length != 2) {
System.err.println("Invalid args");
printUsage();
- }
- else {
+ } else {
transform(input, new File(args[1]));
}
}
@@ -70,13 +67,11 @@ public class XMLConfigurationMigration {
}
private static void recursiveTransform(File root) throws Exception {
- for ( File file : root.listFiles())
- {
+ for (File file : root.listFiles()) {
scanAndTransform(file);
}
}
-
public static void scanAndTransform(File pFile) throws Exception {
try {
for (File f : pFile.listFiles()) {
@@ -93,14 +88,12 @@ public class XMLConfigurationMigration {
file.renameTo(r);
}
}
- }
- catch (Exception e) {
+ } catch (Exception e) {
//continue
}
}
}
- }
- catch (NullPointerException e) {
+ } catch (NullPointerException e) {
System.out.println(pFile.getAbsoluteFile());
}
}
@@ -125,9 +118,7 @@ public class XMLConfigurationMigration {
migration.write(output, properties);
return true;
}
- }
- catch (Exception e)
- {
+ } catch (Exception e) {
System.err.println("Error tranforming document");
e.printStackTrace();
}
@@ -174,8 +165,7 @@ public class XMLConfigurationMigration {
if (addresses.containsKey(addressName)) {
address = addresses.get(addressName);
- }
- else {
+ } else {
address = new Address();
address.setName(addressName);
addresses.put(addressName, address);
[02/17] activemq-artemis git commit: This closes #853
Posted by ma...@apache.org.
This closes #853
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/0f9efa92
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/0f9efa92
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/0f9efa92
Branch: refs/heads/ARTEMIS-780
Commit: 0f9efa9223b4237853c89718549a9bffb3a822fb
Parents: e44c99d 9c16ba2
Author: Clebert Suconic <cl...@apache.org>
Authored: Fri Oct 21 09:33:52 2016 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Fri Oct 21 09:33:52 2016 -0400
----------------------------------------------------------------------
.../impl/ActiveMQServerControlImpl.java | 5 +-
.../paging/cursor/PageSubscriptionCounter.java | 3 +
.../impl/PageSubscriptionCounterImpl.java | 14 +-
.../core/paging/impl/PagingStoreImpl.java | 1 -
.../core/server/ActiveMQMessageBundle.java | 4 +-
.../core/server/ActiveMQServerLogger.java | 5 +
.../artemis/core/server/impl/QueueImpl.java | 15 +-
.../ClusteredMessageCounterTest.java | 318 +++++++++++++++++++
.../management/ActiveMQServerControlTest.java | 9 +-
9 files changed, 359 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
[15/17] activemq-artemis git commit: ARTEMIS-790 Added Configuration
conversion tooL
Posted by ma...@apache.org.
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/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
new file mode 100644
index 0000000..4c3e068
--- /dev/null
+++ b/artemis-tools/src/test/resources/artemis-configuration.xsd
@@ -0,0 +1,2591 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema"
+ xmlns="urn:activemq:core"
+ targetNamespace="urn:activemq:core"
+ attributeFormDefault="unqualified"
+ elementFormDefault="qualified"
+ version="1.0">
+
+ <xsd:element name="core" type="configurationType"/>
+
+ <xsd:complexType name="configurationType">
+ <xsd:all>
+ <xsd:element name="name" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Node name. If set, it will be used in topology notifications.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="resolve-protocols" type="xsd:boolean" default="true" maxOccurs="1"
+ minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ 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.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="persistence-enabled" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ true means that the server will use the file based journal for persistence.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="scheduled-thread-pool-max-size" type="xsd:int" default="5" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Maximum number of threads to use for the scheduled thread pool
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="thread-pool-max-size" type="xsd:int" default="30" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Maximum number of threads to use for the thread pool. -1 means 'no limits'.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="graceful-shutdown-enabled" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ true means that graceful shutdown is enabled
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="graceful-shutdown-timeout" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how long (in ms) to wait for clients to disconnect before shutting down the server
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="security-enabled" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ true means that security is enabled
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="security-invalidation-interval" type="xsd:long" default="10000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how long (in ms) to wait before invalidating the security cache
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-lock-acquisition-timeout" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how long (in ms) to wait to acquire a file lock on the journal
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="wild-card-routing-enabled" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ true means that the server supports wild card routing
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="management-address" type="xsd:string" default="jms.queue.activemq.management" maxOccurs="1"
+ minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the name of the management address to send management messages to. It is prefixed with "jms.queue" so
+ that JMS clients can send messages to it.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="management-notification-address" type="xsd:string" default="activemq.notifications"
+ maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the name of the address that consumers bind to receive management notifications
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="cluster-user" type="xsd:string" default="ACTIVEMQ.CLUSTER.ADMIN.USER" maxOccurs="1"
+ minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Cluster username. It applies to all cluster configurations.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="cluster-password" maxOccurs="1" minOccurs="0" type="xsd:string" default="CHANGE ME!!">
+ <xsd:annotation>
+ <xsd:documentation>
+ Cluster password. It applies to all cluster configurations.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="password-codec" type="xsd:string"
+ default="org.apache.activemq.artemis.utils.DefaultSensitiveStringCodec" maxOccurs="1"
+ minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Class name and its parameters for the Decoder used to decode the masked password. Ignored if
+ mask-password is false. The format of this property is a full qualified class name optionally followed
+ by key/value pairs.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="mask-password" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ This option controls whether passwords in server configuration need be masked. If set to "true" the
+ passwords are masked.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="log-delegate-factory-class-name" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ XXX
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="jmx-management-enabled" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ true means that the management API is available via JMX
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="jmx-domain" type="xsd:string" default="org.apache.activemq" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the JMX domain used to registered ActiveMQ Artemis MBeans in the MBeanServer
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="jmx-use-broker-name" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Whether or not to use the broker name in the JMX properties
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="message-counter-enabled" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ true means that message counters are enabled
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="message-counter-sample-period" type="xsd:long" default="10000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the sample period (in ms) to use for message counters
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="message-counter-max-day-history" type="xsd:int" default="10" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how many days to keep message counter history
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="connection-ttl-override" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ if set, this will override how long (in ms) to keep a connection alive without receiving a ping. -1
+ disables this setting.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="connection-ttl-check-interval" type="xsd:long" default="2000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how often (in ms) to check connections for ttl violation
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="configuration-file-refresh-period" type="xsd:long" default="5000" maxOccurs="1"
+ minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how often (in ms) to check the configuration file for modifications
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="async-connection-execution-enabled" type="xsd:boolean" default="true" maxOccurs="1"
+ minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ should certain incoming packets on the server be handed off to a thread from the thread pool for
+ processing or should they be handled on the remoting thread?
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="transaction-timeout" type="xsd:long" default="300000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how long (in ms) before a transaction can be removed from the resource manager after create time
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="transaction-timeout-scan-period" type="xsd:long" default="1000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how often (in ms) to scan for timeout transactions
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="message-expiry-scan-period" type="xsd:long" default="30000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how often (in ms) to scan for expired messages
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="message-expiry-thread-priority" type="xsd:int" default="3" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the priority of the thread expiring messages
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="id-cache-size" type="xsd:int" default="20000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the size of the cache for pre-creating message ID's
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="persist-id-cache" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ true means that ID's are persisted to the journal
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="remoting-incoming-interceptors" type="class-name-sequenceType" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of <class-name/> elements with the names of classes to use for interceptor incoming
+ remoting packets
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="remoting-outgoing-interceptors" type="class-name-sequenceType" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of <class-name/> elements with the names of classes to use for interceptor outcoming
+ remoting packets
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="persist-delivery-count-before-delivery" type="xsd:boolean" default="false" maxOccurs="1"
+ minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ True means that the delivery count is persisted before delivery. False means that this only happens
+ after a message has been cancelled.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="populate-validated-user" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ true means that the server will add the name of the validated user to messages it sends
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="connectors" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of remoting connectors configurations to create
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element name="connector" type="transportType" maxOccurs="unbounded"/>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element maxOccurs="1" minOccurs="0" name="acceptors">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of remoting acceptors to create
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element name="acceptor" type="transportType" maxOccurs="unbounded"/>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element maxOccurs="1" minOccurs="0" name="broadcast-groups">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of broadcast groups to create
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element ref="broadcast-group" maxOccurs="unbounded" minOccurs="0"/>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element maxOccurs="1" minOccurs="0" name="discovery-groups">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of discovery groups to create
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element maxOccurs="unbounded" minOccurs="0" ref="discovery-group">
+ <xsd:annotation>
+ <xsd:documentation>
+ a discovery group specification element
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element name="diverts" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of diverts to use
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element name="divert" type="divertType" maxOccurs="unbounded" minOccurs="0"/>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+
+ <!-- QUEUES -->
+ <xsd:element name="queues" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of pre configured queues to create
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element name="queue" maxOccurs="unbounded" minOccurs="0">
+ <xsd:complexType>
+ <xsd:all>
+ <xsd:element name="address" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ address for the queue
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element ref="filter" maxOccurs="1" minOccurs="0"/>
+ <xsd:element name="durable" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ whether the queue is durable (persistent)
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:all>
+ <xsd:attribute name="name" type="xsd:ID" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ unique name of this queue
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+ </xsd:element>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element name="bridges" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of bridges to create
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element name="bridge" type="bridgeType" maxOccurs="unbounded" minOccurs="0"/>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element name="ha-policy" type="haPolicyType" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The HA policy of this server
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="cluster-connections" type="clusterConnectionChoiceType" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of cluster connections
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="grouping-handler" type="groupingHandlerType" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Message Group configuration
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="paging-directory" type="xsd:string" default="data/paging" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the directory to store paged messages in
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="bindings-directory" type="xsd:string" default="data/bindings" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the directory to store the persisted bindings to
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="create-bindings-dir" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ true means that the server will create the bindings directory on start up
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="page-max-concurrent-io" type="xsd:int" default="5" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The max number of concurrent reads allowed on paging
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-directory" type="xsd:string" default="data/journal" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the directory to store the journal files in
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="create-journal-dir" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ true means that the journal directory will be created
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-type" default="ASYNCIO" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the type of journal to use
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:simpleType>
+ <xsd:restriction base="xsd:string">
+ <xsd:enumeration value="ASYNCIO"/>
+ <xsd:enumeration value="NIO"/>
+ </xsd:restriction>
+ </xsd:simpleType>
+ </xsd:element>
+
+ <xsd:element name="journal-buffer-timeout" type="xsd:long" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The timeout (in nanoseconds) used to flush internal buffers on the journal. The exact default value
+ depend on whether the journal is ASYNCIO or NIO.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-buffer-size" type="xsd:long" default="501760" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The size of the internal buffer on the journal in KiB.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-sync-transactional" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ if true wait for transaction data to be synchronized to the journal before returning response to
+ client
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-sync-non-transactional" type="xsd:boolean" default="true" maxOccurs="1"
+ minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ if true wait for non transaction data to be synced to the journal before returning response to client.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="log-journal-write-rate" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Whether to log messages about the journal write rate
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-file-size" default="10485760" type="xsd:int" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the size (in bytes) of each journal file
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-min-files" type="xsd:int" default="2" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how many journal files to pre-create
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-pool-files" type="xsd:int" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how many journal files to pre-create
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-compact-percentage" type="xsd:int" default="30" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The percentage of live data on which we consider compacting the journal
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-compact-min-files" type="xsd:int" default="10" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The minimal number of data files before we can start compacting
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="journal-max-io" type="xsd:int" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the maximum number of write requests that can be in the AIO queue at any one time. Default is 500 for
+ AIO and 1 for NIO.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="perf-blast-pages" type="xsd:int" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ XXX Only meant to be used by project developers
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="run-sync-speed-test" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ XXX Only meant to be used by project developers
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="server-dump-interval" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Interval to log server specific information (e.g. memory usage etc)
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="global-max-size" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Global Max Size before all addresses will enter into their Full Policy configured upon messages being
+ produced.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="max-disk-usage" type="xsd:int" default="90" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Max percentage of disk usage before the system blocks or fail clients.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="disk-scan-period" type="xsd:long" default="5000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how often (in ms) to scan the disks for full disks.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="memory-warning-threshold" type="xsd:int" default="25" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Percentage of available memory which will trigger a warning log
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="memory-measure-interval" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ frequency to sample JVM memory in ms (or -1 to disable memory sampling)
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="large-messages-directory" type="xsd:string" default="data/largemessages"
+ maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the directory to store large messages
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="store" type="storeType" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The Store Type used by the server
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="security-settings" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of security settings
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:choice>
+ <xsd:element name="security-setting" maxOccurs="unbounded" minOccurs="0">
+ <xsd:complexType>
+ <xsd:annotation>
+ <xsd:documentation>
+ a permission to add to the matched addresses
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:sequence>
+ <xsd:element name="permission" maxOccurs="unbounded" minOccurs="0">
+ <xsd:complexType>
+ <xsd:attribute name="type" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ the type of permission
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ <xsd:attribute name="roles" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ a comma-separated list of roles to apply the permission to
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+ </xsd:element>
+ </xsd:sequence>
+ <xsd:attribute name="match" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ regular expression for matching security roles against addresses
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+ </xsd:element>
+ <xsd:element name="security-setting-plugin" maxOccurs="1" minOccurs="0">
+ <xsd:complexType>
+ <xsd:annotation>
+ <xsd:documentation>
+ a plugin
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:sequence>
+ <xsd:element name="setting" maxOccurs="unbounded" minOccurs="0">
+ <xsd:complexType>
+ <xsd:attribute name="name" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ the name of the setting
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ <xsd:attribute name="value" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ the value for the setting
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+ </xsd:element>
+ </xsd:sequence>
+ <xsd:attribute name="class-name" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ the name of the plugin class to instantiate
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+ </xsd:element>
+ </xsd:choice>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element name="address-settings" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of address settings
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element ref="address-setting" maxOccurs="unbounded" minOccurs="0"/>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element name="resource-limit-settings" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of resource limit settings
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element ref="resource-limit-setting" maxOccurs="unbounded" minOccurs="0"/>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element name="connector-services" maxOccurs="1" minOccurs="0">
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element name="connector-service" type="connector-serviceType" maxOccurs="unbounded"
+ minOccurs="0"/>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element name="addresses" type="addressesType" maxOccurs="1" minOccurs="0" />
+ </xsd:all>
+ </xsd:complexType>
+
+ <xsd:element name="local-bind-address" type="xsd:string">
+ <xsd:annotation>
+ <xsd:documentation>
+ local bind address that the datagram socket is bound to
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="local-bind-port" type="xsd:int" default="-1">
+ <xsd:annotation>
+ <xsd:documentation>
+ local port to which the datagram socket is bound to
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <!-- BROADCAST GROUP CONFIGURATION -->
+ <xsd:element name="broadcast-group">
+ <xsd:complexType>
+ <xsd:sequence>
+ <!-- XXX these 2 local-* here...-->
+ <xsd:element ref="local-bind-address" maxOccurs="1" minOccurs="0"/>
+ <xsd:element ref="local-bind-port" maxOccurs="1" minOccurs="0"/>
+ <xsd:element name="group-address" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ multicast address to which the data will be broadcast
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="group-port" type="xsd:int" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ UDP port number used for broadcasting
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="broadcast-period" type="xsd:long" default="2000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ period in milliseconds between consecutive broadcasts
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="jgroups-file" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Name of JGroups configuration file. If specified, the server uses JGroups for broadcasting.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="jgroups-channel" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Name of JGroups Channel. If specified, the server uses the named channel for broadcasting.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="connector-ref" type="xsd:string" maxOccurs="unbounded" minOccurs="0"/>
+ </xsd:sequence>
+
+ <xsd:attribute name="name" type="xsd:ID" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ a unique name for the broadcast group
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element name="discovery-group">
+ <xsd:complexType>
+ <xsd:all>
+ <!-- XXX -->
+ <xsd:element name="group-address" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Multicast IP address of the group to listen on
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="group-port" type="xsd:int" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ UDP port number of the multi cast group
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="jgroups-file" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Name of a JGroups configuration file. If specified, the server uses JGroups for discovery.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="jgroups-channel" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Name of a JGroups Channel. If specified, the server uses the named channel for discovery.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="refresh-timeout" type="xsd:int" default="10000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Period the discovery group waits after receiving the last broadcast from a particular server before
+ removing that servers connector pair entry from its list.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element ref="local-bind-address" maxOccurs="1" minOccurs="0"/>
+ <xsd:element ref="local-bind-port" maxOccurs="1" minOccurs="0"/>
+ <xsd:element name="initial-wait-timeout" type="xsd:int" default="10000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ time to wait for an initial broadcast to give us at least one node in the cluster
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:all>
+
+ <xsd:attribute name="name" type="xsd:ID" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ a unique name for the discovery group
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:element name="discovery-group-ref">
+ <xsd:complexType>
+ <xsd:attribute name="discovery-group-name" type="xsd:IDREF"/>
+ </xsd:complexType>
+ </xsd:element>
+
+ <xsd:complexType name="class-name-sequenceType">
+ <xsd:annotation>
+ <xsd:documentation>
+ unlimited sequence of <class-name/>
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:sequence>
+ <xsd:element maxOccurs="unbounded" minOccurs="1" name="class-name" type="xsd:string">
+ <xsd:annotation>
+ <xsd:documentation>
+ the fully qualified name of the interceptor class
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:sequence>
+ </xsd:complexType>
+
+ <xsd:complexType name="paramType">
+ <xsd:attribute name="key" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ Key of a configuration parameter
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ <xsd:attribute name="value" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ Value of a configuration parameter
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+
+ <!-- BRIDGE CONFIGURATION -->
+ <xsd:complexType name="bridgeType">
+ <xsd:sequence>
+ <xsd:element name="queue-name" type="xsd:IDREF" maxOccurs="1" minOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ name of queue that this bridge consumes from
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="forwarding-address" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ address to forward to. If omitted original address is used
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="ha" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ whether this bridge supports fail-over
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element ref="filter" maxOccurs="1" minOccurs="0"/>
+
+ <xsd:element name="transformer-class-name" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ optional name of transformer class
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="min-large-message-size" type="xsd:int" default="102400" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Any message larger than this size is considered a large message (to be sent in
+ chunks)
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="check-period" type="xsd:long" default="30000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The period (in milliseconds) a bridge's client will check if it failed to receive a ping from the
+ server. -1 disables this check.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="connection-ttl" type="xsd:long" default="60000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how long to keep a connection alive in the absence of any data arriving from the client. This should
+ be greater than the ping period.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="retry-interval" type="xsd:long" default="2000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ period (in ms) between successive retries
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="retry-interval-multiplier" type="xsd:double" default="1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ multiplier to apply to successive retry intervals
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="max-retry-interval" type="xsd:long" default="2000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Limit to the retry-interval growth (due to retry-interval-multiplier)
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="initial-connect-attempts" type="xsd:int" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ maximum number of initial connection attempts, -1 means 'no limits'
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="reconnect-attempts" type="xsd:int" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ maximum number of retry attempts, -1 means 'no limits'
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="failover-on-server-shutdown" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ should failover be prompted if target server is cleanly shutdown?
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="use-duplicate-detection" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ should duplicate detection headers be inserted in forwarded messages?
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="confirmation-window-size" type="xsd:int" maxOccurs="1" minOccurs="0" default="1048576">
+ <xsd:annotation>
+ <xsd:documentation>
+ Once the bridge has received this many bytes, it sends a confirmation
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="producer-window-size" type="xsd:int" maxOccurs="1" minOccurs="0" default="-1">
+ <xsd:annotation>
+ <xsd:documentation>
+ Producer flow control
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="user" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ username, if unspecified the cluster-user is used
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="password" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ password, if unspecified the cluster-password is used
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="reconnect-attempts-same-node" default="10" type="xsd:int" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Upon reconnection this configures the number of time the same node on the topology will be retried
+ before reseting the server locator and using the initial connectors
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:choice>
+ <xsd:element name="static-connectors" maxOccurs="1" minOccurs="1">
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element name="connector-ref" type="xsd:string" maxOccurs="unbounded" minOccurs="1"/>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+ <xsd:element name="discovery-group-ref" maxOccurs="1" minOccurs="1">
+ <xsd:complexType>
+ <xsd:attribute name="discovery-group-name" type="xsd:IDREF" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ name of discovery group used by this bridge
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+ </xsd:element>
+ </xsd:choice>
+ </xsd:sequence>
+
+ <xsd:attribute name="name" type="xsd:ID" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ unique name for this bridge
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+
+
+ <!-- CLUSTER CONNECTION CONFIGURATION -->
+
+ <xsd:complexType name="clusterConnectionChoiceType">
+ <xsd:sequence>
+ <xsd:choice maxOccurs="unbounded">
+ <xsd:element name="cluster-connection-uri" type="cluster-connectionUriType"/>
+ <xsd:element name="cluster-connection" type="cluster-connectionType">
+ </xsd:element>
+ </xsd:choice>
+ </xsd:sequence>
+ </xsd:complexType>
+
+ <xsd:complexType name="cluster-connectionUriType">
+ <xsd:attribute name="address" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ uri of the cluster connection
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ <xsd:attribute name="name" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ name of the cluster connection
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+
+ <xsd:complexType name="cluster-connectionType">
+ <xsd:sequence>
+ <xsd:element name="address" type="xsd:string" maxOccurs="1" minOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ name of the address this cluster connection applies to
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="connector-ref" type="xsd:string" maxOccurs="1" minOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ Name of the connector reference to use.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="check-period" type="xsd:long" default="30000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The period (in milliseconds) used to check if the cluster connection has failed to receive pings from
+ another server
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="connection-ttl" type="xsd:long" default="60000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how long to keep a connection alive in the absence of any data arriving from the client
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="min-large-message-size" type="xsd:int" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Messages larger than this are considered large-messages
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="call-timeout" type="xsd:long" default="30000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ How long to wait for a reply
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="retry-interval" type="xsd:long" default="500" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ period (in ms) between successive retries
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="retry-interval-multiplier" type="xsd:double" default="1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ multiplier to apply to the retry-interval
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="max-retry-interval" type="xsd:long" default="2000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Maximum value for retry-interval
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="initial-connect-attempts" type="xsd:int" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ How many attempts should be made to connect initially
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="reconnect-attempts" type="xsd:int" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ How many attempts should be made to reconnect after failure
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="use-duplicate-detection" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ should duplicate detection headers be inserted in forwarded messages?
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="forward-when-no-consumers" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ DEPRECATED: use message-load-balancing-type instead. Select STRICT to mimic
+ forward-when-no-consumers=true
+ and ON_DEMAND to mimic forward-when-no-consumers=false.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="message-load-balancing" default="ON_DEMAND" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how should messages be load balanced between servers in a cluster?
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:simpleType>
+ <xsd:restriction base="xsd:string">
+ <xsd:enumeration value="OFF"/>
+ <xsd:enumeration value="STRICT"/>
+ <xsd:enumeration value="ON_DEMAND"/>
+ </xsd:restriction>
+ </xsd:simpleType>
+ </xsd:element>
+
+ <xsd:element name="max-hops" type="xsd:int" default="1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ maximum number of hops cluster topology is propagated
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="confirmation-window-size" type="xsd:int" default="1048576" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The size (in bytes) of the window used for confirming data from the server connected to.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="producer-window-size" type="xsd:int" maxOccurs="1" minOccurs="0" default="-1">
+ <xsd:annotation>
+ <xsd:documentation>
+ Producer flow control
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="call-failover-timeout" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ How long to wait for a reply if in the middle of a fail-over. -1 means wait forever.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="notification-interval" type="xsd:long" default="1000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how often the cluster connection will notify the cluster of its existence right after joining the
+ cluster
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="notification-attempts" type="xsd:int" default="2" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ how many times this cluster connection will notify the cluster of its existence right after joining
+ the cluster
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="scale-down-connector" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The connector to use for scaling down or when as backup in SCALE_DOWN mode
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:choice>
+ <xsd:element name="static-connectors" maxOccurs="1" minOccurs="0">
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element name="connector-ref" type="xsd:string" maxOccurs="unbounded" minOccurs="0"/>
+ </xsd:sequence>
+ <xsd:attribute name="allow-direct-connections-only" default="false" type="xsd:boolean" use="optional">
+ <xsd:annotation>
+ <xsd:documentation>
+ restricts cluster connections to the listed connector-ref's
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+ </xsd:element>
+ <xsd:element name="discovery-group-ref" maxOccurs="1" minOccurs="0">
+ <xsd:complexType>
+ <xsd:attribute name="discovery-group-name" type="xsd:IDREF" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ XXX -- this is a duplicate...
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+ </xsd:element>
+ </xsd:choice>
+ </xsd:sequence>
+ <xsd:attribute name="name" type="xsd:ID" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ unique name for this cluster connection
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ <xsd:attribute name="uri" type="xsd:string" use="optional">
+ <xsd:annotation>
+ <xsd:documentation>
+ The URI for the cluster connection options
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+
+
+ <xsd:complexType name="cluster-connection-uri-type">
+ <xsd:attribute name="name" type="xsd:ID" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ unique name for this cluster connection
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ <xsd:attribute name="uri" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ The URI for the cluster connection options
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+
+ <!-- DIVERT CONFIGURATION TYPE -->
+ <xsd:complexType name="divertType">
+ <xsd:all>
+ <xsd:element name="transformer-class-name" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ an optional class name of a transformer
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="exclusive" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ whether this is an exclusive divert
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="routing-name" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the routing name for the divert
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="address" type="xsd:string" maxOccurs="1" minOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ the address this divert will divert from
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="forwarding-address" type="xsd:string" maxOccurs="1" minOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ the forwarding address for the divert
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element ref="filter" maxOccurs="1" minOccurs="0"/>
+ </xsd:all>
+
+ <xsd:attribute name="name" type="xsd:ID" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ a unique name for the divert
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+
+ </xsd:complexType>
+
+ <xsd:complexType name="storeType">
+ <xsd:choice>
+ <xsd:element name="file-store" type="fileStoreType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ Use a file based store for peristing journal, paging and large messages
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="database-store" type="databaseStoreType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ Use a database for persisting journal, paging and large messages
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:choice>
+ </xsd:complexType>
+
+ <xsd:complexType name="fileStoreType">
+ </xsd:complexType>
+
+ <xsd:complexType name="databaseStoreType">
+ <xsd:all>
+ <xsd:element name="jdbc-driver-class-name" type="xsd:string" minOccurs="1" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The JDBC Driver class name
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="jdbc-connection-url" type="xsd:string" minOccurs="1" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The JDBC Connection URL e.g. jdbc:mysql://localhost:3306/
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="message-table-name" type="xsd:string" minOccurs="1" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The table name used to store message journal entries
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="bindings-table-name" type="xsd:string" minOccurs="1" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The table name used to store bindings journal entries
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="large-message-table-name" type="xsd:string" minOccurs="1" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The table name used to large message files
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:all>
+ </xsd:complexType>
+
+ <xsd:complexType name="haPolicyType">
+ <xsd:choice>
+ <xsd:element name="live-only" type="haLiveOnlyPolicyType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ A live only server with no HA capabilities apart from scale down.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="replication" type="haReplicationType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ Configuration for a replicated server, either master, slave or colocated.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="shared-store" type="haSharedStoreType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ Configuration for a shared store server, either master, slave or colocated.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:choice>
+ </xsd:complexType>
+
+ <xsd:complexType name="haReplicationType">
+ <xsd:choice>
+ <xsd:element name="master" type="replicatedPolicyType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ A live server configured to replicate.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="slave" type="replicaPolicyType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ A backup server configured to replicate.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="colocated" type="haColocationReplicationType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ a replicated lives server that will allow requests to create colocated replicated backup servers.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:choice>
+ </xsd:complexType>
+
+
+ <xsd:complexType name="haColocationReplicationType">
+ <xsd:all>
+ <xsd:element name="request-backup" type="xsd:boolean" minOccurs="0" maxOccurs="1" default="false">
+ <xsd:annotation>
+ <xsd:documentation>
+ If true then the server will request a backup on another node
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="backup-request-retries" type="xsd:int" minOccurs="0" maxOccurs="1" default="-1">
+ <xsd:annotation>
+ <xsd:documentation>
+ How many times the live server will try to request a backup, -1 means for ever.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="backup-request-retry-interval" type="xsd:long" minOccurs="0" maxOccurs="1" default="5000">
+ <xsd:annotation>
+ <xsd:documentation>
+ How long to wait for retries between attempts to request a backup server.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="max-backups" type="xsd:int" minOccurs="0" maxOccurs="1" default="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ Whether or not this live server will accept backup requests from other live servers.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="backup-port-offset" type="xsd:int" minOccurs="0" maxOccurs="1" default="100">
+ <xsd:annotation>
+ <xsd:documentation>
+ The offset to use for the Connectors and Acceptors when creating a new backup server.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="excludes" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the connectors that shouldn't have their ports offset, typically remote connectors or the
+ connector used in the cluster connection if scalinmg down
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element name="connector-ref" type="xsd:string" maxOccurs="unbounded" minOccurs="1"/>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+ <xsd:element name="master" type="replicatedPolicyType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The configuration for the live replicated server.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="slave" type="replicaPolicyType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The configuration for any slaves created.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:all>
+ </xsd:complexType>
+
+ <xsd:complexType name="haColocationSharedStoreType">
+ <xsd:all>
+ <xsd:element name="request-backup" type="xsd:boolean" minOccurs="0" maxOccurs="1" default="false">
+ <xsd:annotation>
+ <xsd:documentation>
+ If true then the server will request a backup on another node
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="backup-request-retries" type="xsd:int" minOccurs="0" maxOccurs="1" default="-1">
+ <xsd:annotation>
+ <xsd:documentation>
+ How many times the live server will try to request a backup, -1 means for ever.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="backup-request-retry-interval" type="xsd:long" minOccurs="0" maxOccurs="1" default="5000">
+ <xsd:annotation>
+ <xsd:documentation>
+ How long to wait for retries between attempts to request a backup server.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="max-backups" type="xsd:int" minOccurs="0" maxOccurs="1" default="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ Whether or not this live server will accept backup requests from other live servers.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="backup-port-offset" type="xsd:int" minOccurs="0" maxOccurs="1" default="100">
+ <xsd:annotation>
+ <xsd:documentation>
+ The offset to use for the Connectors and Acceptors when creating a new backup server.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="master" type="sharedStoreMasterPolicyType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The configuration for the live shared store server.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="slave" type="sharedStoreSlavePolicyType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The configuration for any shared store backups created.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:all>
+ </xsd:complexType>
+
+ <xsd:complexType name="haSharedStoreType">
+ <xsd:choice>
+ <xsd:element name="master" type="sharedStoreMasterPolicyType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ A shared store live server configuration.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="slave" type="sharedStoreSlavePolicyType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ A shared store backup server configuration.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="colocated" type="haColocationSharedStoreType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ A shared store colocated configuration
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:choice>
+ </xsd:complexType>
+
+ <xsd:complexType name="haLiveOnlyPolicyType">
+ <xsd:all>
+ <xsd:element name="scale-down" type="scaleDownType" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The scale down configuration of this live server.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:all>
+ </xsd:complexType>
+ <xsd:complexType name="replicatedPolicyType">
+ <xsd:all>
+ <xsd:element name="group-name" type="xsd:string" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ used for replication, if set, (remote) backup servers will only pair with live servers with matching
+ group-name
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="cluster-name" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Name of the cluster configuration to use for replication. This setting is only necessary in case you
+ configure multiple cluster connections. It is used by a replicating backups and by live servers that
+ may attempt fail-back.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="check-for-live-server" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Whether to check the cluster for a (live) server using our own server ID when starting
+ up. This option is only necessary for performing 'fail-back' on replicating
+ servers. Strictly speaking this setting only applies to live servers and not to
+ backups.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="initial-replication-sync-timeout" type="xsd:long" default="30000" maxOccurs="1"
+ minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The amount of time to wait for the replica to acknowledge it has received all the necessary data from
+ the replicating server at the final step of the initial replication synchronization process.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:all>
+ </xsd:complexType>
+ <xsd:complexType name="replicaPolicyType">
+ <xsd:all>
+ <xsd:element name="group-name" type="xsd:string" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ used for replication, if set, (remote) backup servers will only pair with live servers with matching
+ group-name
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="cluster-name" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Name of the cluster configuration to use for replication. This setting is only necessary in case you
+ configure multiple cluster connections. It is used by a replicating backups and by live servers that
+ may attempt fail-back.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="max-saved-replicated-journals-size" type="xsd:int" default="2" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ This specifies how many times a replicated backup server can restart after moving its files on start.
+ Once there are this number of backup journal files the server will stop permanently after if fails
+ back.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="scale-down" type="scaleDownType" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ if provided then this backup will scale down rather than becoming live after fail over.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="restart-backup" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Will this server, if a backup, restart once it has been stopped because of failback or scaling down.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="allow-failback" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Whether a server will automatically stop when a another places a request to take over
+ its place. The use case is when a regular server stops and its backup takes over its
+ duties, later the main server restarts and requests the server (the former backup) to
+ stop operating.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="failback-delay" type="xsd:long" default="5000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ DEPRECATED: if we have to start as a replicated server this is the delay to wait before fail-back
+ occurs
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="initial-replication-sync-timeout" type="xsd:long" default="30000" maxOccurs="1"
+ minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ If we have to start as a replicated server this is the amount of time to wait for the replica to
+ acknowledge it has received all the necessary data from the replicating server at the final step
+ of the initial replication synchronization process.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:all>
+ </xsd:complexType>
+ <xsd:complexType name="colocatedReplicaPolicyType">
+ <xsd:all>
+ <xsd:element name="group-name" type="xsd:string" minOccurs="0" maxOccurs="1">
+ <xsd:annotation>
+ <xsd:documentation>
+ used for replication, if set, (remote) backup servers will only pair with live servers with matching
+ group-name
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="cluster-name" type="xsd:string" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Name of the cluster configuration to use for replication. This setting is only necessary in case you
+ configure multiple cluster connections. It is used by a replicating backups and by live servers that
+ may attempt fail-back.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="max-saved-replicated-journals-size" type="xsd:int" default="2" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ This specifies how many times a replicated backup server can restart after moving its files on start.
+ Once there are this number of backup journal files the server will stop permanently after if fails
+ back.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="scale-down" type="scaleDownType" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ if provided then this backup will scale down rather than becoming live after fail over.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ <xsd:element name="restart-backup" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ Will this server, if a backup, restart once it has been stopped because of failback or scaling down.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+ </xsd:all>
+ </xsd:complexType>
+ <xsd:complexType name="sharedStoreMasterPolicyType">
+ <xsd:all>
+ <xsd:element name="failback-delay" type="xsd:long" default="5000" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ DEPRECATED: delay to wait before fail-back occurs on (live's) restar
<TRUNCATED>
[10/17] activemq-artemis git commit: Consolidate RoutingType impls
Posted by ma...@apache.org.
Consolidate RoutingType impls
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/d6e0e6a8
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/d6e0e6a8
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/d6e0e6a8
Branch: refs/heads/ARTEMIS-780
Commit: d6e0e6a809e386c0b41f8d516e9e6e793ccdeb8c
Parents: 938b953
Author: jbertram <jb...@apache.com>
Authored: Fri Oct 21 10:51:29 2016 -0500
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Mon Oct 24 14:33:53 2016 +0100
----------------------------------------------------------------------
.../core/config/CoreAddressConfiguration.java | 8 ++---
.../core/persistence/AddressBindingInfo.java | 4 +--
.../codec/PersistentAddressBindingEncoding.java | 10 +++---
.../artemis/core/server/impl/AddressInfo.java | 33 ++++++++++++++++----
.../core/config/impl/FileConfigurationTest.java | 7 ++---
pom.xml | 1 +
6 files changed, 41 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d6e0e6a8/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
index e01c398..6327f79 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
@@ -21,13 +21,13 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
-import org.apache.activemq.artemis.core.server.impl.AddressInfo.RoutingType;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
public class CoreAddressConfiguration implements Serializable {
private String name = null;
- private RoutingType routingType = null;
+ private AddressInfo.RoutingType routingType = null;
private Integer defaultMaxConsumers = ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers();
@@ -47,11 +47,11 @@ public class CoreAddressConfiguration implements Serializable {
return this;
}
- public RoutingType getRoutingType() {
+ public AddressInfo.RoutingType getRoutingType() {
return routingType;
}
- public CoreAddressConfiguration setRoutingType(RoutingType routingType) {
+ public CoreAddressConfiguration setRoutingType(AddressInfo.RoutingType routingType) {
this.routingType = routingType;
return this;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d6e0e6a8/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/AddressBindingInfo.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/AddressBindingInfo.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/AddressBindingInfo.java
index 4256774..83d37bc 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/AddressBindingInfo.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/AddressBindingInfo.java
@@ -17,7 +17,7 @@
package org.apache.activemq.artemis.core.persistence;
import org.apache.activemq.artemis.api.core.SimpleString;
-import org.apache.activemq.artemis.core.persistence.impl.RoutingType;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
public interface AddressBindingInfo {
@@ -29,6 +29,6 @@ public interface AddressBindingInfo {
SimpleString getUser();
- RoutingType getRoutingType();
+ AddressInfo.RoutingType getRoutingType();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d6e0e6a8/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentAddressBindingEncoding.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentAddressBindingEncoding.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentAddressBindingEncoding.java
index 8aa54e4..9f47362 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentAddressBindingEncoding.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentAddressBindingEncoding.java
@@ -20,7 +20,7 @@ import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.journal.EncodingSupport;
import org.apache.activemq.artemis.core.persistence.AddressBindingInfo;
-import org.apache.activemq.artemis.core.persistence.impl.RoutingType;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.utils.DataConstants;
public class PersistentAddressBindingEncoding implements EncodingSupport, AddressBindingInfo {
@@ -33,7 +33,7 @@ public class PersistentAddressBindingEncoding implements EncodingSupport, Addres
public SimpleString user;
- public RoutingType routingType;
+ public AddressInfo.RoutingType routingType;
public PersistentAddressBindingEncoding() {
}
@@ -55,7 +55,7 @@ public class PersistentAddressBindingEncoding implements EncodingSupport, Addres
public PersistentAddressBindingEncoding(final SimpleString name,
final SimpleString user,
final boolean autoCreated,
- final RoutingType routingType) {
+ final AddressInfo.RoutingType routingType) {
this.name = name;
this.user = user;
this.autoCreated = autoCreated;
@@ -87,7 +87,7 @@ public class PersistentAddressBindingEncoding implements EncodingSupport, Addres
}
@Override
- public RoutingType getRoutingType() {
+ public AddressInfo.RoutingType getRoutingType() {
return routingType;
}
@@ -109,7 +109,7 @@ public class PersistentAddressBindingEncoding implements EncodingSupport, Addres
}
autoCreated = buffer.readBoolean();
- routingType = RoutingType.getType(buffer.readByte());
+ routingType = AddressInfo.RoutingType.getType(buffer.readByte());
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d6e0e6a8/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
index 03c3fa0..4c6ec1f 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
@@ -16,18 +16,13 @@
*/
package org.apache.activemq.artemis.core.server.impl;
-import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.SimpleString;
public class AddressInfo {
- public enum RoutingType {
- MULTICAST, ANYCAST
- }
-
private final SimpleString name;
- private RoutingType routingType = RoutingType.MULTICAST;
+ private RoutingType routingType = RoutingType.Multicast;
private boolean defaultDeleteOnNoConsumers;
@@ -64,4 +59,30 @@ public class AddressInfo {
public SimpleString getName() {
return name;
}
+
+ public enum RoutingType {
+ Multicast, Anycast;
+
+ public byte getType() {
+ switch (this) {
+ case Multicast:
+ return 0;
+ case Anycast:
+ return 1;
+ default:
+ return -1;
+ }
+ }
+
+ public static RoutingType getType(byte type) {
+ switch (type) {
+ case 0:
+ return Multicast;
+ case 1:
+ return Anycast;
+ default:
+ return null;
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d6e0e6a8/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
index ce924c0..f7a0175 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
@@ -52,9 +52,6 @@ import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.junit.Assert;
import org.junit.Test;
-import static org.apache.activemq.artemis.core.server.impl.AddressInfo.RoutingType.ANYCAST;
-import static org.apache.activemq.artemis.core.server.impl.AddressInfo.RoutingType.MULTICAST;
-
public class FileConfigurationTest extends ConfigurationImplTest {
private final String fullConfigurationName = "ConfigurationTest-full-config.xml";
@@ -370,7 +367,7 @@ public class FileConfigurationTest extends ConfigurationImplTest {
// Addr 1
CoreAddressConfiguration addressConfiguration = conf.getAddressConfigurations().get(0);
assertEquals("addr1", addressConfiguration.getName());
- assertEquals(ANYCAST, addressConfiguration.getRoutingType());
+ assertEquals(AddressInfo.RoutingType.Anycast, addressConfiguration.getRoutingType());
assertEquals(2, addressConfiguration.getQueueConfigurations().size());
// Addr 1 Queue 1
@@ -396,7 +393,7 @@ public class FileConfigurationTest extends ConfigurationImplTest {
// Addr 2
addressConfiguration = conf.getAddressConfigurations().get(1);
assertEquals("addr2", addressConfiguration.getName());
- assertEquals(MULTICAST, addressConfiguration.getRoutingType());
+ assertEquals(AddressInfo.RoutingType.Multicast, addressConfiguration.getRoutingType());
assertEquals(2, addressConfiguration.getQueueConfigurations().size());
// Addr 2 Queue 1
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d6e0e6a8/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 1c7d346..f5ace5d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1259,6 +1259,7 @@
<exclude>docs/**/_book/</exclude>
<exclude>**/target/</exclude>
<exclude>**/META-INF/services/*</exclude>
+ <exclude>**/META-INF/MANIFEST.MF</exclude>
<exclude>**/*.iml</exclude>
<exclude>**/*.jceks</exclude>
<exclude>**/*.jks</exclude>
[17/17] activemq-artemis git commit: ARTEMIS-781 add journal record
for address binding
Posted by ma...@apache.org.
ARTEMIS-781 add journal record for address binding
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/938b9539
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/938b9539
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/938b9539
Branch: refs/heads/ARTEMIS-780
Commit: 938b9539071f8ba83df45bd163b54f66dbd99d35
Parents: 5a80eeb
Author: jbertram <jb...@apache.com>
Authored: Mon Oct 10 15:11:50 2016 -0500
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Mon Oct 24 14:33:53 2016 +0100
----------------------------------------------------------------------
.../core/persistence/AddressBindingInfo.java | 34 +++++
.../core/persistence/impl/RoutingType.java | 43 ++++++
.../codec/PersistentAddressBindingEncoding.java | 135 +++++++++++++++++++
3 files changed, 212 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/938b9539/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/AddressBindingInfo.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/AddressBindingInfo.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/AddressBindingInfo.java
new file mode 100644
index 0000000..4256774
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/AddressBindingInfo.java
@@ -0,0 +1,34 @@
+/*
+ * 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.core.persistence;
+
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.persistence.impl.RoutingType;
+
+public interface AddressBindingInfo {
+
+ long getId();
+
+ SimpleString getName();
+
+ boolean isAutoCreated();
+
+ SimpleString getUser();
+
+ RoutingType getRoutingType();
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/938b9539/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/RoutingType.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/RoutingType.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/RoutingType.java
new file mode 100644
index 0000000..329d8e9
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/RoutingType.java
@@ -0,0 +1,43 @@
+/*
+ * 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.core.persistence.impl;
+
+public enum RoutingType {
+ Multicast, Anycast;
+
+ public byte getType() {
+ switch (this) {
+ case Multicast:
+ return 0;
+ case Anycast:
+ return 1;
+ default:
+ return -1;
+ }
+ }
+
+ public static RoutingType getType(byte type) {
+ switch (type) {
+ case 0:
+ return Multicast;
+ case 1:
+ return Anycast;
+ default:
+ return null;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/938b9539/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentAddressBindingEncoding.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentAddressBindingEncoding.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentAddressBindingEncoding.java
new file mode 100644
index 0000000..8aa54e4
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentAddressBindingEncoding.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.core.persistence.impl.journal.codec;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.journal.EncodingSupport;
+import org.apache.activemq.artemis.core.persistence.AddressBindingInfo;
+import org.apache.activemq.artemis.core.persistence.impl.RoutingType;
+import org.apache.activemq.artemis.utils.DataConstants;
+
+public class PersistentAddressBindingEncoding implements EncodingSupport, AddressBindingInfo {
+
+ public long id;
+
+ public SimpleString name;
+
+ public boolean autoCreated;
+
+ public SimpleString user;
+
+ public RoutingType routingType;
+
+ public PersistentAddressBindingEncoding() {
+ }
+
+ @Override
+ public String toString() {
+ return "PersistentAddressBindingEncoding [id=" + id +
+ ", name=" +
+ name +
+ ", user=" +
+ user +
+ ", autoCreated=" +
+ autoCreated +
+ ", routingType=" +
+ routingType +
+ "]";
+ }
+
+ public PersistentAddressBindingEncoding(final SimpleString name,
+ final SimpleString user,
+ final boolean autoCreated,
+ final RoutingType routingType) {
+ this.name = name;
+ this.user = user;
+ this.autoCreated = autoCreated;
+ this.routingType = routingType;
+ }
+
+ @Override
+ public long getId() {
+ return id;
+ }
+
+ public void setId(final long id) {
+ this.id = id;
+ }
+
+ @Override
+ public SimpleString getName() {
+ return name;
+ }
+
+ @Override
+ public SimpleString getUser() {
+ return user;
+ }
+
+ @Override
+ public boolean isAutoCreated() {
+ return autoCreated;
+ }
+
+ @Override
+ public RoutingType getRoutingType() {
+ return routingType;
+ }
+
+ @Override
+ public void decode(final ActiveMQBuffer buffer) {
+ name = buffer.readSimpleString();
+
+ String metadata = buffer.readNullableSimpleString().toString();
+ if (metadata != null) {
+ String[] elements = metadata.split(";");
+ for (String element : elements) {
+ String[] keyValuePair = element.split("=");
+ if (keyValuePair.length == 2) {
+ if (keyValuePair[0].equals("user")) {
+ user = SimpleString.toSimpleString(keyValuePair[1]);
+ }
+ }
+ }
+ }
+
+ autoCreated = buffer.readBoolean();
+ routingType = RoutingType.getType(buffer.readByte());
+ }
+
+ @Override
+ public void encode(final ActiveMQBuffer buffer) {
+ buffer.writeSimpleString(name);
+ buffer.writeNullableSimpleString(createMetadata());
+ buffer.writeBoolean(autoCreated);
+ buffer.writeByte(routingType.getType());
+ }
+
+ @Override
+ public int getEncodeSize() {
+ return SimpleString.sizeofString(name) + DataConstants.SIZE_BOOLEAN +
+ SimpleString.sizeofNullableString(createMetadata()) +
+ DataConstants.SIZE_BYTE;
+ }
+
+ private SimpleString createMetadata() {
+ StringBuilder metadata = new StringBuilder();
+ metadata.append("user=").append(user).append(";");
+ return SimpleString.toSimpleString(metadata.toString());
+ }
+}
[04/17] activemq-artemis git commit: This closes #856
Posted by ma...@apache.org.
This closes #856
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/8d4f507c
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/8d4f507c
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/8d4f507c
Branch: refs/heads/ARTEMIS-780
Commit: 8d4f507cb83dd91b8ba6acc35a895367fa305ee1
Parents: 0f9efa9 a074f9f
Author: jbertram <jb...@apache.com>
Authored: Fri Oct 21 19:58:16 2016 -0500
Committer: jbertram <jb...@apache.com>
Committed: Fri Oct 21 19:58:16 2016 -0500
----------------------------------------------------------------------
.../api/core/management/QueueControl.java | 8 ++-
.../api/jms/management/JMSQueueControl.java | 6 ++
.../management/impl/JMSQueueControlImpl.java | 5 ++
.../core/management/impl/QueueControlImpl.java | 12 ++++
.../core/persistence/QueueBindingInfo.java | 7 ++
.../artemis/core/persistence/QueueStatus.java | 46 ++++++++++++
.../core/persistence/StorageManager.java | 11 +++
.../journal/AbstractJournalStorageManager.java | 66 ++++++++++++++++-
.../impl/journal/DescribeJournal.java | 4 ++
.../impl/journal/JournalRecordIds.java | 2 +
.../codec/PersistentQueueBindingEncoding.java | 18 +++++
.../impl/journal/codec/QueueEncoding.java | 3 +-
.../impl/journal/codec/QueueStatusEncoding.java | 75 ++++++++++++++++++++
.../impl/nullpm/NullStorageManager.java | 11 +++
.../activemq/artemis/core/server/Queue.java | 15 ++++
.../server/impl/PostOfficeJournalLoader.java | 13 +++-
.../artemis/core/server/impl/QueueImpl.java | 41 +++++++++++
.../impl/ScheduledDeliveryHandlerTest.java | 13 ++++
.../transaction/impl/TransactionImplTest.java | 11 +++
.../management/JMSQueueControlUsingJMSTest.java | 5 ++
.../management/QueueControlUsingCoreTest.java | 5 ++
.../server/QueuePeristPauseTest.java | 60 ++++++++++++++++
.../unit/core/postoffice/impl/FakeQueue.java | 15 ++++
23 files changed, 448 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
[05/17] activemq-artemis git commit: ARTEMIS-816 Log warning during
boot if no dead letter/expire address is configured
Posted by ma...@apache.org.
ARTEMIS-816 Log warning during boot if no dead letter/expire address is configured
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/e95e4f77
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/e95e4f77
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/e95e4f77
Branch: refs/heads/ARTEMIS-780
Commit: e95e4f775b1258e33cafa3059755efaa9f9f825e
Parents: 8d4f507
Author: bayern39 <ja...@163.com>
Authored: Sat Oct 22 11:28:03 2016 +0800
Committer: bayern39 <ja...@163.com>
Committed: Sat Oct 22 11:30:55 2016 +0800
----------------------------------------------------------------------
.../artemis/core/server/ActiveMQServerLogger.java | 10 ++++++++++
.../activemq/artemis/core/server/impl/QueueImpl.java | 10 ++++++++++
2 files changed, 20 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e95e4f77/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
index 24432a3..51aa57b 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
@@ -1025,6 +1025,16 @@ public interface ActiveMQServerLogger extends BasicLogger {
void errorStartingReplication(BackupReplicationStartFailedMessage.BackupRegistrationProblem problem);
@LogMessage(level = Logger.Level.WARN)
+ @Message(id = 222165, value = "No Dead Letter Address configured for queue {0} in AddressSettings",
+ format = Message.Format.MESSAGE_FORMAT)
+ void AddressSettingsNoDLA(SimpleString name);
+
+ @LogMessage(level = Logger.Level.WARN)
+ @Message(id = 222166, value = "No Expiry Address configured for queue {0} in AddressSettings",
+ format = Message.Format.MESSAGE_FORMAT)
+ void AddressSettingsNoExpiryAddress(SimpleString name);
+
+ @LogMessage(level = Logger.Level.WARN)
@Message(id = 222167, value = "Group Binding not available so deleting {0} groups from {1}, groups will be bound to another node",
format = Message.Format.MESSAGE_FORMAT)
void groupingQueueRemoved(int size, SimpleString clusterName);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e95e4f77/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index d30544f..b70fe8d 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -2989,10 +2989,20 @@ public class QueueImpl implements Queue {
public void onChange() {
AddressSettings settings = addressSettingsRepository.getMatch(address.toString());
configureExpiry(settings);
+ checkDeadLetterAddressAndExpiryAddress(settings);
configureSlowConsumerReaper(settings);
}
}
+ private void checkDeadLetterAddressAndExpiryAddress(final AddressSettings settings) {
+ if (settings.getDeadLetterAddress() == null) {
+ ActiveMQServerLogger.LOGGER.AddressSettingsNoDLA(name);
+ }
+ if (settings.getExpiryAddress() == null) {
+ ActiveMQServerLogger.LOGGER.AddressSettingsNoExpiryAddress(name);
+ }
+ }
+
private final class SlowConsumerReaperRunnable implements Runnable {
private final SlowConsumerPolicy policy;
[12/17] activemq-artemis git commit: Added ANYCAST routing to local
queues
Posted by ma...@apache.org.
Added ANYCAST routing to local queues
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/6d41d37e
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/6d41d37e
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/6d41d37e
Branch: refs/heads/ARTEMIS-780
Commit: 6d41d37e9b2f069eb3e8c935c428fd2876730f46
Parents: c2d4f3e
Author: Martyn Taylor <mt...@redhat.com>
Authored: Mon Oct 24 14:27:00 2016 +0100
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Mon Oct 24 14:33:53 2016 +0100
----------------------------------------------------------------------
.../artemis/core/postoffice/AddressManager.java | 2 +
.../artemis/core/postoffice/PostOffice.java | 2 +
.../core/postoffice/impl/BindingsImpl.java | 1 +
.../core/postoffice/impl/LocalQueueBinding.java | 9 +-
.../core/postoffice/impl/PostOfficeImpl.java | 5 +
.../postoffice/impl/SimpleAddressManager.java | 15 ++
.../artemis/core/server/ActiveMQServer.java | 2 +-
.../core/server/impl/ActiveMQServerImpl.java | 12 +-
.../artemis/core/server/impl/AddressInfo.java | 12 +-
.../server/impl/PostOfficeJournalLoader.java | 3 +-
.../core/server/impl/QueueFactoryImpl.java | 8 +
.../core/config/impl/FileConfigurationTest.java | 4 +-
.../integration/addressing/AddressingTest.java | 240 ++++++++++++++++++-
.../integration/client/HangConsumerTest.java | 2 +-
.../jms/client/TopicCleanupTest.java | 2 +-
.../core/server/impl/fakes/FakePostOffice.java | 5 +
16 files changed, 300 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
index 5519822..1cf1a07 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
@@ -54,6 +54,8 @@ public interface AddressManager {
AddressInfo addAddressInfo(AddressInfo addressInfo);
+ AddressInfo addOrUpdateAddressInfo(AddressInfo addressInfo);
+
AddressInfo removeAddressInfo(SimpleString address);
AddressInfo getAddressInfo(SimpleString address);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
index f719966..7902352 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
@@ -45,6 +45,8 @@ public interface PostOffice extends ActiveMQComponent {
AddressInfo addAddressInfo(AddressInfo addressInfo);
+ AddressInfo addOrUpdateAddressInfo(AddressInfo addressInfo);
+
AddressInfo removeAddressInfo(SimpleString address);
AddressInfo getAddressInfo(SimpleString address);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
index e5df737..6be0311 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
@@ -262,6 +262,7 @@ public final class BindingsImpl implements Bindings {
boolean routed = false;
for (Binding binding : exclusiveBindings) {
+
if (binding.getFilter() == null || binding.getFilter().match(message)) {
binding.getBindable().route(message, context);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/LocalQueueBinding.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/LocalQueueBinding.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/LocalQueueBinding.java
index 2a6d9c5..2921388 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/LocalQueueBinding.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/LocalQueueBinding.java
@@ -24,10 +24,11 @@ import org.apache.activemq.artemis.core.server.Bindable;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.RoutingContext;
import org.apache.activemq.artemis.core.server.ServerMessage;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
public class LocalQueueBinding implements QueueBinding {
- private final SimpleString address;
+ private final AddressInfo address;
private final Queue queue;
@@ -37,7 +38,7 @@ public class LocalQueueBinding implements QueueBinding {
private final SimpleString clusterName;
- public LocalQueueBinding(final SimpleString address, final Queue queue, final SimpleString nodeID) {
+ public LocalQueueBinding(final AddressInfo address, final Queue queue, final SimpleString nodeID) {
this.address = address;
this.queue = queue;
@@ -61,7 +62,7 @@ public class LocalQueueBinding implements QueueBinding {
@Override
public SimpleString getAddress() {
- return address;
+ return address.getName();
}
@Override
@@ -76,7 +77,7 @@ public class LocalQueueBinding implements QueueBinding {
@Override
public SimpleString getRoutingName() {
- return name;
+ return (address.getRoutingType() == AddressInfo.RoutingType.MULTICAST) ? name : address.getName();
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
index 9b7ed0c..6c654bf 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
@@ -425,6 +425,11 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
}
@Override
+ public AddressInfo addOrUpdateAddressInfo(AddressInfo addressInfo) {
+ return addressManager.addOrUpdateAddressInfo(addressInfo);
+ }
+
+ @Override
public AddressInfo removeAddressInfo(SimpleString address) {
return addressManager.removeAddressInfo(address);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
index 2994f9e..969a1a9 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
@@ -188,6 +188,21 @@ public class SimpleAddressManager implements AddressManager {
}
@Override
+ public AddressInfo addOrUpdateAddressInfo(AddressInfo addressInfo) {
+ AddressInfo from = addAddressInfo(addressInfo);
+ return (from == null) ? addressInfo : updateAddressInfo(from, addressInfo);
+ }
+
+ private AddressInfo updateAddressInfo(AddressInfo from, AddressInfo to) {
+ synchronized (from) {
+ from.setRoutingType(to.getRoutingType());
+ from.setDefaultMaxConsumers(to.getDefaultMaxConsumers());
+ from.setDefaultDeleteOnNoConsumers(to.isDefaultDeleteOnNoConsumers());
+ return from;
+ }
+ }
+
+ @Override
public AddressInfo removeAddressInfo(SimpleString address) {
return addressInfoMap.remove(address);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/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 8bac182..1fbcdb1 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
@@ -411,7 +411,7 @@ public interface ActiveMQServer extends ActiveMQComponent {
void removeClientConnection(String clientId);
- AddressInfo addAddressInfo(AddressInfo addressInfo);
+ AddressInfo createOrUpdateAddressInfo(AddressInfo addressInfo);
AddressInfo removeAddressInfo(SimpleString address);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/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 513c431..b3b6b5e 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
@@ -2050,7 +2050,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
info.setDefaultDeleteOnNoConsumers(config.getDefaultDeleteOnNoConsumers());
info.setDefaultMaxConsumers(config.getDefaultMaxConsumers());
- addAddressInfo(info);
+ createOrUpdateAddressInfo(info);
deployQueuesFromListCoreQueueConfiguration(config.getQueueConfigurations());
}
}
@@ -2154,8 +2154,8 @@ public class ActiveMQServerImpl implements ActiveMQServer {
}
@Override
- public AddressInfo addAddressInfo(AddressInfo addressInfo) {
- return postOffice.addAddressInfo(addressInfo);
+ public AddressInfo createOrUpdateAddressInfo(AddressInfo addressInfo) {
+ return postOffice.addOrUpdateAddressInfo(addressInfo);
}
@Override
@@ -2165,7 +2165,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
@Override
public AddressInfo getAddressInfo(SimpleString address) {
- return postOffice.removeAddressInfo(address);
+ return postOffice.getAddressInfo(address);
}
private Queue createQueue(final SimpleString addressName,
@@ -2201,15 +2201,13 @@ public class ActiveMQServerImpl implements ActiveMQServer {
final QueueConfig queueConfig = queueConfigBuilder.filter(filter).pagingManager(pagingManager).user(user).durable(durable).temporary(temporary).autoCreated(autoCreated).build();
final Queue queue = queueFactory.createQueueWith(queueConfig);
- addAddressInfo(new AddressInfo(queue.getAddress()));
-
if (transientQueue) {
queue.setConsumersRefCount(new TransientQueueManagerImpl(this, queue.getName()));
} else if (queue.isAutoCreated()) {
queue.setConsumersRefCount(new AutoCreatedQueueManagerImpl(this.getJMSQueueDeleter(), queue.getName()));
}
- final QueueBinding localQueueBinding = new LocalQueueBinding(queue.getAddress(), queue, nodeManager.getNodeId());
+ final QueueBinding localQueueBinding = new LocalQueueBinding(getAddressInfo(queue.getAddress()), queue, nodeManager.getNodeId());
if (queue.isDurable()) {
storageManager.addQueueBinding(txID, localQueueBinding);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
index 4c6ec1f..1449107 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
@@ -22,7 +22,7 @@ public class AddressInfo {
private final SimpleString name;
- private RoutingType routingType = RoutingType.Multicast;
+ private RoutingType routingType = RoutingType.MULTICAST;
private boolean defaultDeleteOnNoConsumers;
@@ -61,13 +61,13 @@ public class AddressInfo {
}
public enum RoutingType {
- Multicast, Anycast;
+ MULTICAST, ANYCAST;
public byte getType() {
switch (this) {
- case Multicast:
+ case MULTICAST:
return 0;
- case Anycast:
+ case ANYCAST:
return 1;
default:
return -1;
@@ -77,9 +77,9 @@ public class AddressInfo {
public static RoutingType getType(byte type) {
switch (type) {
case 0:
- return Multicast;
+ return MULTICAST;
case 1:
- return Anycast;
+ return ANYCAST;
default:
return null;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
index 9a8ae74..71c5b2b 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
@@ -155,7 +155,8 @@ public class PostOfficeJournalLoader implements JournalLoader {
}
}
- final Binding binding = new LocalQueueBinding(queue.getAddress(), queue, nodeManager.getNodeId());
+ final Binding binding = new LocalQueueBinding(postOffice.getAddressInfo(queue.getAddress()), queue, nodeManager.getNodeId());
+
queues.put(queue.getID(), queue);
postOffice.addBinding(binding);
managementService.registerAddress(queue.getAddress());
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
index 5686c7b..3678553 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
@@ -68,6 +68,10 @@ public class QueueFactoryImpl implements QueueFactory {
@Override
public Queue createQueueWith(final QueueConfig config) {
+
+ // Add default address info if one doesn't exist
+ postOffice.addAddressInfo(new AddressInfo(config.address()));
+
final AddressSettings addressSettings = addressSettingsRepository.getMatch(config.address().toString());
final Queue queue;
if (addressSettings.isLastValueQueue()) {
@@ -89,6 +93,10 @@ public class QueueFactoryImpl implements QueueFactory {
final boolean durable,
final boolean temporary,
final boolean autoCreated) {
+
+ // Add default address info if one doesn't exist
+ postOffice.addAddressInfo(new AddressInfo(address));
+
AddressSettings addressSettings = addressSettingsRepository.getMatch(address.toString());
Queue queue;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
index f7a0175..46f3958 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
@@ -367,7 +367,7 @@ public class FileConfigurationTest extends ConfigurationImplTest {
// Addr 1
CoreAddressConfiguration addressConfiguration = conf.getAddressConfigurations().get(0);
assertEquals("addr1", addressConfiguration.getName());
- assertEquals(AddressInfo.RoutingType.Anycast, addressConfiguration.getRoutingType());
+ assertEquals(AddressInfo.RoutingType.ANYCAST, addressConfiguration.getRoutingType());
assertEquals(2, addressConfiguration.getQueueConfigurations().size());
// Addr 1 Queue 1
@@ -393,7 +393,7 @@ public class FileConfigurationTest extends ConfigurationImplTest {
// Addr 2
addressConfiguration = conf.getAddressConfigurations().get(1);
assertEquals("addr2", addressConfiguration.getName());
- assertEquals(AddressInfo.RoutingType.Multicast, addressConfiguration.getRoutingType());
+ assertEquals(AddressInfo.RoutingType.MULTICAST, addressConfiguration.getRoutingType());
assertEquals(2, addressConfiguration.getQueueConfigurations().size());
// Addr 2 Queue 1
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
index 43d6071..2e0fda4 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
@@ -16,6 +16,244 @@
*/
package org.apache.activemq.artemis.tests.integration.addressing;
-public class AddressingTest {
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.api.core.TransportConfiguration;
+import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
+import org.apache.activemq.artemis.api.core.client.ClientConsumer;
+import org.apache.activemq.artemis.api.core.client.ClientMessage;
+import org.apache.activemq.artemis.api.core.client.ClientProducer;
+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.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.junit.Before;
+import org.junit.Test;
+
+public class AddressingTest extends ActiveMQTestBase {
+
+ private ActiveMQServer server;
+
+ private ClientSessionFactory sessionFactory;
+
+ @Before
+ public void setup() throws Exception {
+ server = createServer(true);
+ server.start();
+
+ server.waitForActivation(10, TimeUnit.SECONDS);
+
+ ServerLocator sl = ActiveMQClient.createServerLocatorWithoutHA(new TransportConfiguration(INVM_CONNECTOR_FACTORY));
+ sessionFactory = sl.createSessionFactory();
+
+ addSessionFactory(sessionFactory);
+ }
+
+ @Test
+ public void testMulticastRouting() throws Exception {
+
+ SimpleString sendAddress = new SimpleString("test.address");
+
+ List<String> testAddresses = Arrays.asList("test.address", "test.#", "test.*");
+
+ for (String consumeAddress : testAddresses) {
+
+ // For each address, create 2 Queues with the same address, assert both queues receive message
+
+ AddressInfo addressInfo = new AddressInfo(new SimpleString(consumeAddress));
+ addressInfo.setRoutingType(AddressInfo.RoutingType.MULTICAST);
+
+ server.createOrUpdateAddressInfo(addressInfo);
+ Queue q1 = server.createQueue(new SimpleString(consumeAddress), new SimpleString(consumeAddress + ".1"), null, true, false);
+ Queue q2 = server.createQueue(new SimpleString(consumeAddress), new SimpleString(consumeAddress + ".2"), null, true, false);
+
+ ClientSession session = sessionFactory.createSession();
+ session.start();
+
+ ClientConsumer consumer1 = session.createConsumer(q1.getName());
+ ClientConsumer consumer2 = session.createConsumer(q2.getName());
+
+ ClientProducer producer = session.createProducer(sendAddress);
+ ClientMessage m = session.createMessage(ClientMessage.TEXT_TYPE, true);
+ m.getBodyBuffer().writeString("TestMessage");
+
+ producer.send(m);
+
+ assertNotNull(consumer1.receive(2000));
+ assertNotNull(consumer2.receive(2000));
+
+ q1.deleteQueue();
+ q2.deleteQueue();
+
+ System.out.println(consumeAddress);
+ }
+ }
+
+ @Test
+ public void testAnycastRouting() throws Exception {
+
+ SimpleString sendAddress = new SimpleString("test.address");
+
+ List<String> testAddresses = Arrays.asList("test.address", "test.#", "test.*");
+
+ for (String consumeAddress : testAddresses) {
+
+ // For each address, create 2 Queues with the same address, assert one queue receive message
+
+ AddressInfo addressInfo = new AddressInfo(new SimpleString(consumeAddress));
+ addressInfo.setRoutingType(AddressInfo.RoutingType.ANYCAST);
+
+ server.createOrUpdateAddressInfo(addressInfo);
+ Queue q1 = server.createQueue(new SimpleString(consumeAddress), new SimpleString(consumeAddress + ".1"), null, true, false);
+ Queue q2 = server.createQueue(new SimpleString(consumeAddress), new SimpleString(consumeAddress + ".2"), null, true, false);
+
+ ClientSession session = sessionFactory.createSession();
+ session.start();
+
+ ClientConsumer consumer1 = session.createConsumer(q1.getName());
+ ClientConsumer consumer2 = session.createConsumer(q2.getName());
+
+ ClientProducer producer = session.createProducer(sendAddress);
+ ClientMessage m = session.createMessage(ClientMessage.TEXT_TYPE, true);
+
+ m.getBodyBuffer().writeString("TestMessage");
+
+ producer.send(m);
+
+ int count = 0;
+ count = (consumer1.receive(1000) == null) ? count : count + 1;
+ count = (consumer2.receive(1000) == null) ? count : count + 1;
+ assertEquals(1, count);
+
+ q1.deleteQueue();
+ q2.deleteQueue();
+
+ System.out.println(consumeAddress);
+ }
+ }
+
+ @Test
+ public void testAnycastRoutingRoundRobin() throws Exception {
+
+ SimpleString address = new SimpleString("test.address");
+ AddressInfo addressInfo = new AddressInfo(address);
+ addressInfo.setRoutingType(AddressInfo.RoutingType.ANYCAST);
+
+ server.createOrUpdateAddressInfo(addressInfo);
+ Queue q1 = server.createQueue(address, address.concat(".1"), null, true, false);
+ Queue q2 = server.createQueue(address, address.concat(".2"), null, true, false);
+ Queue q3 = server.createQueue(address, address.concat(".3"), null, true, false);
+
+ ClientSession session = sessionFactory.createSession();
+ session.start();
+
+ ClientProducer producer = session.createProducer(address);
+
+ ClientConsumer consumer1 = session.createConsumer(q1.getName());
+ ClientConsumer consumer2 = session.createConsumer(q2.getName());
+ ClientConsumer consumer3 = session.createConsumer(q3.getName());
+ List<ClientConsumer> consumers = new ArrayList<>(Arrays.asList(new ClientConsumer[] {consumer1, consumer2, consumer3}));
+
+ List<String> messages = new ArrayList<>();
+ messages.add("Message1");
+ messages.add("Message2");
+ messages.add("Message3");
+
+ ClientMessage clientMessage;
+ for (String message : messages) {
+ clientMessage = session.createMessage(true);
+ clientMessage.getBodyBuffer().writeString(message);
+ producer.send(clientMessage);
+ }
+
+ String m;
+ for (ClientConsumer consumer : consumers) {
+ clientMessage = consumer.receive(1000);
+ m = clientMessage.getBodyBuffer().readString();
+ messages.remove(m);
+ }
+
+ assertTrue(messages.isEmpty());
+
+ // Check we don't receive more messages
+ int count = 0;
+ for (ClientConsumer consumer : consumers) {
+ count = (consumer.receive(1000) == null) ? count : count + 1;
+ }
+ assertEquals(0, count);
+ }
+
+
+
+ @Test
+ public void testMulticastRoutingBackwardsCompat() throws Exception {
+
+ SimpleString sendAddress = new SimpleString("test.address");
+
+ List<String> testAddresses = Arrays.asList("test.address", "test.#", "test.*");
+
+ for (String consumeAddress : testAddresses) {
+
+ // For each address, create 2 Queues with the same address, assert both queues receive message
+ Queue q1 = server.createQueue(new SimpleString(consumeAddress), new SimpleString(consumeAddress + ".1"), null, true, false);
+ Queue q2 = server.createQueue(new SimpleString(consumeAddress), new SimpleString(consumeAddress + ".2"), null, true, false);
+
+ ClientSession session = sessionFactory.createSession();
+ session.start();
+
+ ClientConsumer consumer1 = session.createConsumer(q1.getName());
+ ClientConsumer consumer2 = session.createConsumer(q2.getName());
+
+ ClientProducer producer = session.createProducer(sendAddress);
+ ClientMessage m = session.createMessage(ClientMessage.TEXT_TYPE, true);
+ m.getBodyBuffer().writeString("TestMessage");
+
+ producer.send(m);
+
+ assertNotNull(consumer1.receive(2000));
+ assertNotNull(consumer2.receive(2000));
+
+ q1.deleteQueue();
+ q2.deleteQueue();
+
+ System.out.println(consumeAddress);
+ }
+ }
+
+ @Test
+ public void testDeleteQueueOnNoConsumersTrue() {
+ fail("Not Implemented");
+ }
+
+ @Test
+ public void testDeleteQueueOnNoConsumersFalse() {
+ fail("Not Implemented");
+ }
+
+ @Test
+ public void testLimitOnMaxConsumers() {
+ fail("Not Implemented");
+ }
+
+ @Test
+ public void testUnlimitedMaxConsumers() {
+ fail("Not Implemented");
+ }
+
+ @Test
+ public void testDefaultMaxConsumersFromAddress() {
+ fail("Not Implemented");
+ }
+
+ @Test
+ public void testDefaultDeleteOnNoConsumersFromAddress() {
+ fail("Not Implemented");
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java
index 83d28a1..2fd5915 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java
@@ -353,7 +353,7 @@ public class HangConsumerTest extends ActiveMQTestBase {
long txID = server.getStorageManager().generateID();
// Forcing a situation where the server would unexpectedly create a duplicated queue. The server should still start normally
- LocalQueueBinding newBinding = new LocalQueueBinding(QUEUE, new QueueImpl(queueID, QUEUE, QUEUE, null, null, true, false, false, null, null, null, null, null), server.getNodeID());
+ LocalQueueBinding newBinding = new LocalQueueBinding(server.getAddressInfo(QUEUE), new QueueImpl(queueID, QUEUE, QUEUE, null, null, true, false, false, null, null, null, null, null), server.getNodeID());
server.getStorageManager().addQueueBinding(txID, newBinding);
server.getStorageManager().commitBindings(txID);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TopicCleanupTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TopicCleanupTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TopicCleanupTest.java
index 280596a..ec279ee 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TopicCleanupTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TopicCleanupTest.java
@@ -83,7 +83,7 @@ public class TopicCleanupTest extends JMSTestBase {
final Queue queue = new QueueImpl(storage.generateID(), SimpleString.toSimpleString("jms.topic.topic"), SimpleString.toSimpleString("jms.topic.topic"), FilterImpl.createFilter(ActiveMQServerImpl.GENERIC_IGNORED_FILTER), null, true, false, false, server.getScheduledPool(), server.getPostOffice(), storage, server.getAddressSettingsRepository(), server.getExecutorFactory().getExecutor());
- LocalQueueBinding binding = new LocalQueueBinding(queue.getAddress(), queue, server.getNodeID());
+ LocalQueueBinding binding = new LocalQueueBinding(server.getAddressInfo(queue.getAddress()), queue, server.getNodeID());
storage.addQueueBinding(txid, binding);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/6d41d37e/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
----------------------------------------------------------------------
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
index 9424fc3..512f0f2 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
@@ -65,6 +65,11 @@ public class FakePostOffice implements PostOffice {
return null;
}
+ @Override
+ public AddressInfo addOrUpdateAddressInfo(AddressInfo addressInfo) {
+ return null;
+ }
+
@Override
public AddressInfo removeAddressInfo(SimpleString address) {
[11/17] activemq-artemis git commit: ARTEMIS-813 Load AddressInfo
into AddressManager
Posted by ma...@apache.org.
ARTEMIS-813 Load AddressInfo into AddressManager
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/b825f62a
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/b825f62a
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/b825f62a
Branch: refs/heads/ARTEMIS-780
Commit: b825f62a3611b738ff37000dcfc2ddbc8ed60af0
Parents: 8c51ed7
Author: Martyn Taylor <mt...@redhat.com>
Authored: Wed Oct 19 19:16:31 2016 +0100
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Mon Oct 24 14:33:53 2016 +0100
----------------------------------------------------------------------
.../core/config/CoreAddressConfiguration.java | 6 +-
.../deployers/impl/FileConfigurationParser.java | 3 +-
.../artemis/core/postoffice/AddressManager.java | 7 ++
.../artemis/core/postoffice/PostOffice.java | 8 +++
.../core/postoffice/impl/PostOfficeImpl.java | 16 +++++
.../postoffice/impl/SimpleAddressManager.java | 18 ++++++
.../artemis/core/server/ActiveMQServer.java | 13 +++-
.../core/server/impl/ActiveMQServerImpl.java | 49 ++++++++++++--
.../artemis/core/server/impl/AddressInfo.java | 67 ++++++++++++++++++++
.../core/config/impl/FileConfigurationTest.java | 5 +-
.../integration/addressing/AddressingTest.java | 21 ++++++
.../core/server/impl/fakes/FakePostOffice.java | 17 +++++
12 files changed, 213 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
index cb6d43f..e01c398 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
@@ -21,14 +21,10 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo.RoutingType;
public class CoreAddressConfiguration implements Serializable {
- public enum RoutingType {
- MULTICAST,
- ANYCAST
- }
-
private String name = null;
private RoutingType routingType = null;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/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 fc045b3..2dccb03 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
@@ -64,6 +64,7 @@ import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.core.server.SecuritySettingPlugin;
import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
import org.apache.activemq.artemis.core.server.group.impl.GroupingHandlerConfiguration;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.settings.impl.ResourceLimitSettings;
@@ -895,7 +896,7 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
CoreAddressConfiguration addressConfiguration = new CoreAddressConfiguration();
addressConfiguration.setName(name)
- .setRoutingType(CoreAddressConfiguration.RoutingType.valueOf(routingType.toUpperCase()));
+ .setRoutingType(AddressInfo.RoutingType.valueOf(routingType.toUpperCase()));
NodeList children = node.getChildNodes();
for (int j = 0; j < children.getLength(); j++) {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
index efc1297..5519822 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
@@ -20,6 +20,7 @@ import java.util.Map;
import java.util.Set;
import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.transaction.Transaction;
/**
@@ -50,4 +51,10 @@ public interface AddressManager {
Map<SimpleString, Binding> getBindings();
Set<SimpleString> getAddresses();
+
+ AddressInfo addAddressInfo(AddressInfo addressInfo);
+
+ AddressInfo removeAddressInfo(SimpleString address);
+
+ AddressInfo getAddressInfo(SimpleString address);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
index 4c0c4b0..f719966 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
@@ -27,6 +27,7 @@ import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.QueueCreator;
import org.apache.activemq.artemis.core.server.RoutingContext;
import org.apache.activemq.artemis.core.server.ServerMessage;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.transaction.Transaction;
/**
@@ -42,6 +43,12 @@ import org.apache.activemq.artemis.core.transaction.Transaction;
*/
public interface PostOffice extends ActiveMQComponent {
+ AddressInfo addAddressInfo(AddressInfo addressInfo);
+
+ AddressInfo removeAddressInfo(SimpleString address);
+
+ AddressInfo getAddressInfo(SimpleString address);
+
void addBinding(Binding binding) throws Exception;
Binding removeBinding(SimpleString uniqueName, Transaction tx, boolean deleteData) throws Exception;
@@ -113,4 +120,5 @@ public interface PostOffice extends ActiveMQComponent {
Set<SimpleString> getAddresses();
+
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
index cc06603..9b7ed0c 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
@@ -69,6 +69,7 @@ import org.apache.activemq.artemis.core.server.RouteContextList;
import org.apache.activemq.artemis.core.server.RoutingContext;
import org.apache.activemq.artemis.core.server.ServerMessage;
import org.apache.activemq.artemis.core.server.group.GroupingHandler;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl;
import org.apache.activemq.artemis.core.server.impl.ServerMessageImpl;
import org.apache.activemq.artemis.core.server.management.ManagementService;
@@ -418,6 +419,21 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
// PostOffice implementation -----------------------------------------------
+ @Override
+ public AddressInfo addAddressInfo(AddressInfo addressInfo) {
+ return addressManager.addAddressInfo(addressInfo);
+ }
+
+ @Override
+ public AddressInfo removeAddressInfo(SimpleString address) {
+ return addressManager.removeAddressInfo(address);
+ }
+
+ @Override
+ public AddressInfo getAddressInfo(SimpleString addressName) {
+ return addressManager.getAddressInfo(addressName);
+ }
+
// TODO - needs to be synchronized to prevent happening concurrently with activate()
// (and possible removeBinding and other methods)
// Otherwise can have situation where createQueue comes in before failover, then failover occurs
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
index 9f26b0b..2994f9e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
@@ -30,6 +30,7 @@ import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.Bindings;
import org.apache.activemq.artemis.core.postoffice.BindingsFactory;
import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.jboss.logging.Logger;
@@ -40,6 +41,8 @@ public class SimpleAddressManager implements AddressManager {
private static final Logger logger = Logger.getLogger(Page.class);
+ private final ConcurrentMap<SimpleString, AddressInfo> addressInfoMap = new ConcurrentHashMap<>();
+
/**
* HashMap<Address, Binding>
*/
@@ -178,4 +181,19 @@ public class SimpleAddressManager implements AddressManager {
return prevBindings != null;
}
+
+ @Override
+ public AddressInfo addAddressInfo(AddressInfo addressInfo) {
+ return addressInfoMap.putIfAbsent(addressInfo.getName(), addressInfo);
+ }
+
+ @Override
+ public AddressInfo removeAddressInfo(SimpleString address) {
+ return addressInfoMap.remove(address);
+ }
+
+ @Override
+ public AddressInfo getAddressInfo(SimpleString addressName) {
+ return addressInfoMap.get(addressName);
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/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 477f839..8bac182 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
@@ -39,6 +39,7 @@ import org.apache.activemq.artemis.core.server.cluster.ClusterManager;
import org.apache.activemq.artemis.core.server.cluster.ha.HAPolicy;
import org.apache.activemq.artemis.core.server.group.GroupingHandler;
import org.apache.activemq.artemis.core.server.impl.Activation;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.server.impl.ConnectorsService;
import org.apache.activemq.artemis.core.server.management.ManagementService;
import org.apache.activemq.artemis.core.server.reload.ReloadManager;
@@ -375,10 +376,12 @@ public interface ActiveMQServer extends ActiveMQComponent {
void stop(boolean failoverOnServerShutdown) throws Exception;
+ AddressInfo getAddressInfo(SimpleString address);
+
/*
- * add a ProtocolManagerFactory to be used. Note if @see Configuration#isResolveProtocols is tur then this factory will
- * replace any factories with the same protocol
- * */
+ * add a ProtocolManagerFactory to be used. Note if @see Configuration#isResolveProtocols is tur then this factory will
+ * replace any factories with the same protocol
+ * */
void addProtocolManagerFactory(ProtocolManagerFactory factory);
/*
@@ -407,4 +410,8 @@ public interface ActiveMQServer extends ActiveMQComponent {
boolean addClientConnection(String clientId, boolean unique);
void removeClientConnection(String clientId);
+
+ AddressInfo addAddressInfo(AddressInfo addressInfo);
+
+ AddressInfo removeAddressInfo(SimpleString address);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/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 98abce0..513c431 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
@@ -54,6 +54,7 @@ import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryImpl;
import org.apache.activemq.artemis.core.config.BridgeConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.ConfigurationUtils;
+import org.apache.activemq.artemis.core.config.CoreAddressConfiguration;
import org.apache.activemq.artemis.core.config.CoreQueueConfiguration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.config.StoreConfiguration;
@@ -1961,6 +1962,9 @@ public class ActiveMQServerImpl implements ActiveMQServer {
// Deploy the rest of the stuff
+ // Deploy predefined addresses
+ deployAddressesFromConfiguration();
+
// Deploy any predefined queues
deployQueuesFromConfiguration();
@@ -2039,11 +2043,26 @@ public class ActiveMQServerImpl implements ActiveMQServer {
}
}
- private void deployQueuesFromConfiguration() throws Exception {
- for (CoreQueueConfiguration config : configuration.getQueueConfigurations()) {
+ private void deployAddressesFromConfiguration() throws Exception {
+ for (CoreAddressConfiguration config : configuration.getAddressConfigurations()) {
+ AddressInfo info = new AddressInfo(SimpleString.toSimpleString(config.getName()));
+ info.setRoutingType(config.getRoutingType());
+ info.setDefaultDeleteOnNoConsumers(config.getDefaultDeleteOnNoConsumers());
+ info.setDefaultMaxConsumers(config.getDefaultMaxConsumers());
+
+ addAddressInfo(info);
+ deployQueuesFromListCoreQueueConfiguration(config.getQueueConfigurations());
+ }
+ }
+
+ 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);
}
}
+ private void deployQueuesFromConfiguration() throws Exception {
+ deployQueuesFromListCoreQueueConfiguration(configuration.getQueueConfigurations());
+ }
private void checkForPotentialOOMEInAddressConfiguration() {
long totalMaxSizeBytes = 0;
@@ -2134,7 +2153,22 @@ public class ActiveMQServerImpl implements ActiveMQServer {
}
}
- private Queue createQueue(final SimpleString address,
+ @Override
+ public AddressInfo addAddressInfo(AddressInfo addressInfo) {
+ return postOffice.addAddressInfo(addressInfo);
+ }
+
+ @Override
+ public AddressInfo removeAddressInfo(SimpleString address) {
+ return postOffice.removeAddressInfo(address);
+ }
+
+ @Override
+ public AddressInfo getAddressInfo(SimpleString address) {
+ return postOffice.removeAddressInfo(address);
+ }
+
+ private Queue createQueue(final SimpleString addressName,
final SimpleString queueName,
final SimpleString filterString,
final SimpleString user,
@@ -2143,6 +2177,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
final boolean ignoreIfExists,
final boolean transientQueue,
final boolean autoCreated) throws Exception {
+
final QueueBinding binding = (QueueBinding) postOffice.getBinding(queueName);
if (binding != null) {
if (ignoreIfExists) {
@@ -2158,14 +2193,16 @@ public class ActiveMQServerImpl implements ActiveMQServer {
final long queueID = storageManager.generateID();
final QueueConfig.Builder queueConfigBuilder;
- if (address == null) {
+ if (addressName == null) {
queueConfigBuilder = QueueConfig.builderWith(queueID, queueName);
} else {
- queueConfigBuilder = QueueConfig.builderWith(queueID, queueName, address);
-
+ queueConfigBuilder = QueueConfig.builderWith(queueID, queueName, addressName);
}
final QueueConfig queueConfig = queueConfigBuilder.filter(filter).pagingManager(pagingManager).user(user).durable(durable).temporary(temporary).autoCreated(autoCreated).build();
final Queue queue = queueFactory.createQueueWith(queueConfig);
+
+ addAddressInfo(new AddressInfo(queue.getAddress()));
+
if (transientQueue) {
queue.setConsumersRefCount(new TransientQueueManagerImpl(this, queue.getName()));
} else if (queue.isAutoCreated()) {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
new file mode 100644
index 0000000..03c3fa0
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
@@ -0,0 +1,67 @@
+/*
+ * 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.core.server.impl;
+
+import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
+import org.apache.activemq.artemis.api.core.SimpleString;
+
+public class AddressInfo {
+
+ public enum RoutingType {
+ MULTICAST, ANYCAST
+ }
+
+ private final SimpleString name;
+
+ private RoutingType routingType = RoutingType.MULTICAST;
+
+ private boolean defaultDeleteOnNoConsumers;
+
+ private int defaultMaxConsumers;
+
+ public AddressInfo(SimpleString name) {
+ this.name = name;
+ }
+
+ public RoutingType getRoutingType() {
+ return routingType;
+ }
+
+ public void setRoutingType(RoutingType routingType) {
+ this.routingType = routingType;
+ }
+
+ public boolean isDefaultDeleteOnNoConsumers() {
+ return defaultDeleteOnNoConsumers;
+ }
+
+ public void setDefaultDeleteOnNoConsumers(boolean defaultDeleteOnNoConsumers) {
+ this.defaultDeleteOnNoConsumers = defaultDeleteOnNoConsumers;
+ }
+
+ public int getDefaultMaxConsumers() {
+ return defaultMaxConsumers;
+ }
+
+ public void setDefaultMaxConsumers(int defaultMaxConsumers) {
+ this.defaultMaxConsumers = defaultMaxConsumers;
+ }
+
+ public SimpleString getName() {
+ return name;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
index d004a22..ce924c0 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
@@ -46,13 +46,14 @@ import org.apache.activemq.artemis.core.security.Role;
import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.core.server.SecuritySettingPlugin;
import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.server.impl.LegacyLDAPSecuritySettingPlugin;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.junit.Assert;
import org.junit.Test;
-import static org.apache.activemq.artemis.core.config.CoreAddressConfiguration.RoutingType.ANYCAST;
-import static org.apache.activemq.artemis.core.config.CoreAddressConfiguration.RoutingType.MULTICAST;
+import static org.apache.activemq.artemis.core.server.impl.AddressInfo.RoutingType.ANYCAST;
+import static org.apache.activemq.artemis.core.server.impl.AddressInfo.RoutingType.MULTICAST;
public class FileConfigurationTest extends ConfigurationImplTest {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
new file mode 100644
index 0000000..43d6071
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/addressing/AddressingTest.java
@@ -0,0 +1,21 @@
+/*
+ * 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.addressing;
+
+public class AddressingTest {
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/b825f62a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
----------------------------------------------------------------------
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
index 6f23e34..9424fc3 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
@@ -33,6 +33,7 @@ import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.QueueCreator;
import org.apache.activemq.artemis.core.server.RoutingContext;
import org.apache.activemq.artemis.core.server.ServerMessage;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.server.impl.MessageReferenceImpl;
import org.apache.activemq.artemis.core.transaction.Transaction;
@@ -60,6 +61,22 @@ public class FakePostOffice implements PostOffice {
}
@Override
+ public AddressInfo addAddressInfo(AddressInfo addressInfo) {
+ return null;
+ }
+
+
+ @Override
+ public AddressInfo removeAddressInfo(SimpleString address) {
+ return null;
+ }
+
+ @Override
+ public AddressInfo getAddressInfo(SimpleString addressName) {
+ return null;
+ }
+
+ @Override
public void addBinding(final Binding binding) throws Exception {
}
[14/17] activemq-artemis git commit: ARTEMIS-790 Added Configuration
conversion tooL
Posted by ma...@apache.org.
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/artemis-tools/src/test/resources/artemis-server.xsd
----------------------------------------------------------------------
diff --git a/artemis-tools/src/test/resources/artemis-server.xsd b/artemis-tools/src/test/resources/artemis-server.xsd
new file mode 100644
index 0000000..1e2e816
--- /dev/null
+++ b/artemis-tools/src/test/resources/artemis-server.xsd
@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<xsd:schema xmlns="urn:activemq"
+ targetNamespace="urn:activemq"
+ xmlns:xsd="http://www.w3.org/2001/XMLSchema"
+ attributeFormDefault="unqualified"
+ elementFormDefault="qualified"
+ version="1.0">
+ <xsd:element name="configuration">
+ <xsd:annotation>
+ <xsd:documentation>
+ Root element for a document specifying the configuration
+ of a single "standalone" server that does not operate
+ as part of a domain.
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:choice minOccurs="1" maxOccurs="unbounded">
+ <xsd:any namespace="##other">
+ <xsd:annotation>
+ <xsd:documentation>A profile declaration may include configuration
+ elements from other namespaces for the subsystems that make up the profile.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:any>
+ </xsd:choice>
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+</xsd:schema>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/artemis-tools/src/test/resources/broker.xml
----------------------------------------------------------------------
diff --git a/artemis-tools/src/test/resources/broker.xml b/artemis-tools/src/test/resources/broker.xml
new file mode 100644
index 0000000..bd33d59
--- /dev/null
+++ b/artemis-tools/src/test/resources/broker.xml
@@ -0,0 +1,64 @@
+<!--
+ 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.
+-->
+<configuration xmlns="urn:activemq"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="urn:activemq /schema/artemis-server.xsd">
+ <jms xmlns="urn:activemq:jms">
+ <!--the queue used by the example-->
+ <queue name="exampleQueue"/>
+ </jms>
+
+ <core xmlns="urn:activemq:core">
+
+ <persistence-enabled>false</persistence-enabled>
+ <!-- Connectors -->
+
+ <connectors>
+ <connector name="in-vm">vm://0</connector>
+ </connectors>
+
+ <acceptors>
+ <acceptor name="in-vm">vm://0</acceptor>
+ </acceptors>
+
+ <!-- Other config -->
+
+ <security-settings>
+ <!--security for example queue-->
+ <security-setting match="jms.queue.exampleQueue">
+ <permission type="createDurableQueue" roles="guest"/>
+ <permission type="deleteDurableQueue" roles="guest"/>
+ <permission type="createNonDurableQueue" roles="guest"/>
+ <permission type="deleteNonDurableQueue" roles="guest"/>
+ <permission type="consume" roles="guest"/>
+ <permission type="send" roles="guest"/>
+ </security-setting>
+ </security-settings>
+
+ <queues>
+ <queue name="foo">
+ <address>bar</address>
+ </queue>
+ <queue name="bar">
+ <address>afar</address>
+ <durable>true</durable>
+ <filter string="name='car'" />
+ </queue>
+ </queues>
+ </core>
+
+</configuration>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/artemis-tools/src/test/resources/replace/broker.xml
----------------------------------------------------------------------
diff --git a/artemis-tools/src/test/resources/replace/broker.xml b/artemis-tools/src/test/resources/replace/broker.xml
new file mode 100644
index 0000000..bd33d59
--- /dev/null
+++ b/artemis-tools/src/test/resources/replace/broker.xml
@@ -0,0 +1,64 @@
+<!--
+ 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.
+-->
+<configuration xmlns="urn:activemq"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="urn:activemq /schema/artemis-server.xsd">
+ <jms xmlns="urn:activemq:jms">
+ <!--the queue used by the example-->
+ <queue name="exampleQueue"/>
+ </jms>
+
+ <core xmlns="urn:activemq:core">
+
+ <persistence-enabled>false</persistence-enabled>
+ <!-- Connectors -->
+
+ <connectors>
+ <connector name="in-vm">vm://0</connector>
+ </connectors>
+
+ <acceptors>
+ <acceptor name="in-vm">vm://0</acceptor>
+ </acceptors>
+
+ <!-- Other config -->
+
+ <security-settings>
+ <!--security for example queue-->
+ <security-setting match="jms.queue.exampleQueue">
+ <permission type="createDurableQueue" roles="guest"/>
+ <permission type="deleteDurableQueue" roles="guest"/>
+ <permission type="createNonDurableQueue" roles="guest"/>
+ <permission type="deleteNonDurableQueue" roles="guest"/>
+ <permission type="consume" roles="guest"/>
+ <permission type="send" roles="guest"/>
+ </security-setting>
+ </security-settings>
+
+ <queues>
+ <queue name="foo">
+ <address>bar</address>
+ </queue>
+ <queue name="bar">
+ <address>afar</address>
+ <durable>true</durable>
+ <filter string="name='car'" />
+ </queue>
+ </queues>
+ </core>
+
+</configuration>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/artemis-tools/src/test/resources/replace/broker2.xml
----------------------------------------------------------------------
diff --git a/artemis-tools/src/test/resources/replace/broker2.xml b/artemis-tools/src/test/resources/replace/broker2.xml
new file mode 100644
index 0000000..bd33d59
--- /dev/null
+++ b/artemis-tools/src/test/resources/replace/broker2.xml
@@ -0,0 +1,64 @@
+<!--
+ 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.
+-->
+<configuration xmlns="urn:activemq"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="urn:activemq /schema/artemis-server.xsd">
+ <jms xmlns="urn:activemq:jms">
+ <!--the queue used by the example-->
+ <queue name="exampleQueue"/>
+ </jms>
+
+ <core xmlns="urn:activemq:core">
+
+ <persistence-enabled>false</persistence-enabled>
+ <!-- Connectors -->
+
+ <connectors>
+ <connector name="in-vm">vm://0</connector>
+ </connectors>
+
+ <acceptors>
+ <acceptor name="in-vm">vm://0</acceptor>
+ </acceptors>
+
+ <!-- Other config -->
+
+ <security-settings>
+ <!--security for example queue-->
+ <security-setting match="jms.queue.exampleQueue">
+ <permission type="createDurableQueue" roles="guest"/>
+ <permission type="deleteDurableQueue" roles="guest"/>
+ <permission type="createNonDurableQueue" roles="guest"/>
+ <permission type="deleteNonDurableQueue" roles="guest"/>
+ <permission type="consume" roles="guest"/>
+ <permission type="send" roles="guest"/>
+ </security-setting>
+ </security-settings>
+
+ <queues>
+ <queue name="foo">
+ <address>bar</address>
+ </queue>
+ <queue name="bar">
+ <address>afar</address>
+ <durable>true</durable>
+ <filter string="name='car'" />
+ </queue>
+ </queues>
+ </core>
+
+</configuration>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index d040375..1c7d346 100644
--- a/pom.xml
+++ b/pom.xml
@@ -56,6 +56,7 @@
<module>integration/activemq-vertx-integration</module>
<module>integration/artemis-cdi-integration</module>
<module>artemis-distribution</module>
+ <module>artemis-tools</module>
<module>tests</module>
<module>artemis-features</module>
</modules>
[16/17] activemq-artemis git commit: ARTEMIS-790 Added Configuration
conversion tooL
Posted by ma...@apache.org.
ARTEMIS-790 Added Configuration conversion tooL
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/5a80eeb5
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/5a80eeb5
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/5a80eeb5
Branch: refs/heads/ARTEMIS-780
Commit: 5a80eeb598d3e7b668805084d7657722a5b39554
Parents: b825f62
Author: Martyn Taylor <mt...@redhat.com>
Authored: Fri Oct 21 10:51:21 2016 +0100
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Mon Oct 24 14:33:53 2016 +0100
----------------------------------------------------------------------
artemis-tools/pom.xml | 56 +
.../config/XMLConfigurationMigration.java | 237 ++
.../migrate/config/addressing/Address.java | 73 +
.../tools/migrate/config/addressing/Queue.java | 50 +
.../src/main/resources/META-INF/MANIFEST.MF | 2 +
.../config/XMLConfigurationMigrationTest.java | 47 +
.../test/resources/artemis-configuration.xsd | 2591 ++++++++++++++++++
.../src/test/resources/artemis-server.xsd | 46 +
artemis-tools/src/test/resources/broker.xml | 64 +
.../src/test/resources/replace/broker.xml | 64 +
.../src/test/resources/replace/broker2.xml | 64 +
pom.xml | 1 +
12 files changed, 3295 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/artemis-tools/pom.xml
----------------------------------------------------------------------
diff --git a/artemis-tools/pom.xml b/artemis-tools/pom.xml
new file mode 100644
index 0000000..9ce22fd
--- /dev/null
+++ b/artemis-tools/pom.xml
@@ -0,0 +1,56 @@
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>artemis-pom</artifactId>
+ <version>1.5.0-SNAPSHOT</version>
+ </parent>
+
+ <name>ActiveMQ Artemis Tools</name>
+ <groupId>org.apache.activemq.tools</groupId>
+ <artifactId>artemis-tools</artifactId>
+ <packaging>jar</packaging>
+
+ <properties>
+ <activemq.basedir>${project.basedir}/..</activemq.basedir>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <configuration>
+ <archive>
+ <manifestFile>src/main/resources/META-INF/MANIFEST.MF</manifestFile>
+ </archive>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigration.java
----------------------------------------------------------------------
diff --git a/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigration.java b/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigration.java
new file mode 100644
index 0000000..56833ea
--- /dev/null
+++ b/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigration.java
@@ -0,0 +1,237 @@
+/*
+ * 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.tools.migrate.config;
+
+import javax.xml.parsers.DocumentBuilder;
+import javax.xml.parsers.DocumentBuilderFactory;
+import javax.xml.transform.OutputKeys;
+import javax.xml.transform.Transformer;
+import javax.xml.transform.TransformerException;
+import javax.xml.transform.TransformerFactory;
+import javax.xml.transform.dom.DOMSource;
+import javax.xml.transform.stream.StreamResult;
+import javax.xml.xpath.XPath;
+import javax.xml.xpath.XPathConstants;
+import javax.xml.xpath.XPathFactory;
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.activemq.artemis.tools.migrate.config.addressing.Address;
+import org.apache.activemq.artemis.tools.migrate.config.addressing.Queue;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
+
+public class XMLConfigurationMigration {
+
+ private static XMLConfigurationMigration migration;
+
+ private final Document document;
+
+ public static void main(String[] args) throws Exception {
+
+ if (args.length == 0) {
+ System.err.println("Invalid args");
+ printUsage();
+ }
+ else {
+ File input = new File(args[0]);
+ if (input.isDirectory()) {
+ System.out.println("Scanning directory: " + input.getAbsolutePath());
+ recursiveTransform(input);
+ }
+ else {
+ if (args.length != 2) {
+ System.err.println("Invalid args");
+ printUsage();
+ }
+ else {
+ transform(input, new File(args[1]));
+ }
+ }
+ }
+ }
+
+ private static void recursiveTransform(File root) throws Exception {
+ for ( File file : root.listFiles())
+ {
+ scanAndTransform(file);
+ }
+ }
+
+
+ public static void scanAndTransform(File pFile) throws Exception {
+ try {
+ for (File f : pFile.listFiles()) {
+ if (f.isDirectory()) {
+ scanAndTransform(f);
+ } else {
+ try {
+ if (f.getName().endsWith("xml")) {
+ File file = new File(f.getAbsolutePath() + ".new");
+ if (transform(f, file)) {
+ File r = new File(f.getAbsolutePath());
+
+ f.renameTo(new File(f.getAbsolutePath() + ".bk"));
+ file.renameTo(r);
+ }
+ }
+ }
+ catch (Exception e) {
+ //continue
+ }
+ }
+ }
+ }
+ catch (NullPointerException e) {
+ System.out.println(pFile.getAbsoluteFile());
+ }
+ }
+
+ public static void printUsage() {
+ System.out.println("Please specify a directory to scan, or input and output file");
+ }
+
+ public static boolean transform(File input, File output) throws Exception {
+
+ migration = new XMLConfigurationMigration(input);
+ try {
+ if (!input.exists()) {
+ System.err.println("Input file not found: " + input);
+ }
+
+ if (migration.convertQueuesToAddresses()) {
+ Properties properties = new Properties();
+ properties.put(OutputKeys.INDENT, "yes");
+ properties.put("{http://xml.apache.org/xslt}indent-amount", "3");
+ properties.put(OutputKeys.ENCODING, "UTF-8");
+ migration.write(output, properties);
+ return true;
+ }
+ }
+ catch (Exception e)
+ {
+ System.err.println("Error tranforming document");
+ e.printStackTrace();
+ }
+ return false;
+ }
+
+ public XMLConfigurationMigration(File input) throws Exception {
+ DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
+ factory.setIgnoringElementContentWhitespace(true);
+
+ DocumentBuilder db = factory.newDocumentBuilder();
+ this.document = db.parse(input);
+ }
+
+ public boolean convertQueuesToAddresses() throws Exception {
+
+ Map<String, Address> addresses = new HashMap<>();
+
+ String xPathQueues = "/configuration/core/queues";
+ String xPathQueue = "/configuration/core/queues/queue";
+ String xPathAttrName = "@name";
+ String xPathAddress = "address";
+ String xPathFilter = "filter/@string";
+ String xPathDurable = "durable";
+
+ XPath xPath = XPathFactory.newInstance().newXPath();
+
+ NodeList xpathResult = (NodeList) xPath.evaluate(xPathQueue, document, XPathConstants.NODESET);
+ if (xpathResult == null || xpathResult.getLength() == 0) {
+ // doesn't require change
+ return false;
+ }
+
+ for (int i = 0; i < xpathResult.getLength(); i++) {
+ Node queueNode = xpathResult.item(i);
+
+ Queue queue = new Queue();
+ queue.setName(xPath.evaluate(xPathAttrName, queueNode, XPathConstants.STRING).toString());
+ queue.setDurable(xPath.evaluate(xPathDurable, queueNode, XPathConstants.STRING).toString());
+ queue.setFilter(xPath.evaluate(xPathFilter, queueNode, XPathConstants.STRING).toString());
+
+ String addressName = xPath.evaluate(xPathAddress, queueNode, XPathConstants.STRING).toString();
+ Address address;
+
+ if (addresses.containsKey(addressName)) {
+ address = addresses.get(addressName);
+ }
+ else {
+ address = new Address();
+ address.setName(addressName);
+ addresses.put(addressName, address);
+ }
+ address.getQueues().add(queue);
+ }
+
+ Node queues = ((Node) xPath.evaluate(xPathQueues, document, XPathConstants.NODE));
+
+ if (queues != null) {
+ Node core = queues.getParentNode();
+ core.removeChild(queues);
+
+ Element a = document.createElement("addresses");
+ for (Address addr : addresses.values()) {
+ Element eAddr = document.createElement("address");
+ eAddr.setAttribute("name", addr.getName());
+ eAddr.setAttribute("type", addr.getRoutingType());
+
+ if (addr.getQueues().size() > 0) {
+ Element eQueues = document.createElement("queues");
+ for (Queue queue : addr.getQueues()) {
+ Element eQueue = document.createElement("queue");
+ eQueue.setAttribute("name", queue.getName());
+ eQueue.setAttribute("max-consumers", addr.getDefaultMaxConsumers());
+ eQueue.setAttribute("delete-on-no-consumers", addr.getDefaultDeleteOnNoConsumers());
+
+ if (queue.getDurable() != null && !queue.getDurable().isEmpty()) {
+ Element eDurable = document.createElement("durable");
+ eDurable.setTextContent(queue.getDurable());
+ eQueue.appendChild(eDurable);
+ }
+
+ if (queue.getFilter() != null && !queue.getFilter().isEmpty()) {
+ Element eFilter = document.createElement("filter");
+ eFilter.setAttribute("string", queue.getFilter());
+ eQueue.appendChild(eFilter);
+ }
+
+ eQueues.appendChild(eQueue);
+ }
+ eAddr.appendChild(eQueues);
+ }
+ a.appendChild(eAddr);
+ }
+ core.appendChild(a);
+ }
+
+ document.normalize();
+ return true;
+ }
+
+ public void write(File output, Properties outputProperties) throws TransformerException {
+ Transformer transformer = TransformerFactory.newInstance().newTransformer();
+ transformer.setOutputProperties(outputProperties);
+ StreamResult streamResult = new StreamResult(output);
+ transformer.transform(new DOMSource(document), streamResult);
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/addressing/Address.java
----------------------------------------------------------------------
diff --git a/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/addressing/Address.java b/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/addressing/Address.java
new file mode 100644
index 0000000..c483c75
--- /dev/null
+++ b/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/addressing/Address.java
@@ -0,0 +1,73 @@
+/*
+ * 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.tools.migrate.config.addressing;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class Address {
+
+ private String name;
+
+ private String routingType = "multicast";
+
+ private String defaultMaxConsumers = "-1";
+
+ private String defaultDeleteOnNoConsumers = "false";
+
+ private List<Queue> queues = new ArrayList<>();
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getRoutingType() {
+ return routingType;
+ }
+
+ public void setRoutingType(String routingType) {
+ this.routingType = routingType;
+ }
+
+ public String getDefaultMaxConsumers() {
+ return defaultMaxConsumers;
+ }
+
+ public void setDefaultMaxConsumers(String defaultMaxConsumers) {
+ this.defaultMaxConsumers = defaultMaxConsumers;
+ }
+
+ public String getDefaultDeleteOnNoConsumers() {
+ return defaultDeleteOnNoConsumers;
+ }
+
+ public void setDefaultDeleteOnNoConsumers(String defaultDeleteOnNoConsumers) {
+ this.defaultDeleteOnNoConsumers = defaultDeleteOnNoConsumers;
+ }
+
+ public List<Queue> getQueues() {
+ return queues;
+ }
+
+ public void setQueues(List<Queue> queues) {
+ this.queues = queues;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/addressing/Queue.java
----------------------------------------------------------------------
diff --git a/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/addressing/Queue.java b/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/addressing/Queue.java
new file mode 100644
index 0000000..f88b8ef
--- /dev/null
+++ b/artemis-tools/src/main/java/org/apache/activemq/artemis/tools/migrate/config/addressing/Queue.java
@@ -0,0 +1,50 @@
+/*
+ * 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.tools.migrate.config.addressing;
+
+public class Queue {
+
+ String name;
+
+ String filter;
+
+ String durable;
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getFilter() {
+ return filter;
+ }
+
+ public void setFilter(String filter) {
+ this.filter = filter;
+ }
+
+ public String getDurable() {
+ return durable;
+ }
+
+ public void setDurable(String durable) {
+ this.durable = durable;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/artemis-tools/src/main/resources/META-INF/MANIFEST.MF
----------------------------------------------------------------------
diff --git a/artemis-tools/src/main/resources/META-INF/MANIFEST.MF b/artemis-tools/src/main/resources/META-INF/MANIFEST.MF
new file mode 100644
index 0000000..72543bf
--- /dev/null
+++ b/artemis-tools/src/main/resources/META-INF/MANIFEST.MF
@@ -0,0 +1,2 @@
+Manifest-Version: 1.0
+Main-Class: org.apache.activemq.artemis.tools.migrate.config.XMLConfigurationMigration
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a80eeb5/artemis-tools/src/test/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigrationTest.java
----------------------------------------------------------------------
diff --git a/artemis-tools/src/test/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigrationTest.java b/artemis-tools/src/test/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigrationTest.java
new file mode 100644
index 0000000..f68e9c2
--- /dev/null
+++ b/artemis-tools/src/test/java/org/apache/activemq/artemis/tools/migrate/config/XMLConfigurationMigrationTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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.tools.migrate.config;
+
+import javax.xml.transform.OutputKeys;
+import java.io.File;
+import java.util.Properties;
+
+import org.junit.Test;
+
+public class XMLConfigurationMigrationTest {
+
+ @Test
+ public void testQueuesReplacedWithAddresses() throws Exception {
+ File brokerXml = new File(this.getClass().getClassLoader().getResource("broker.xml").toURI());
+ XMLConfigurationMigration tool = new XMLConfigurationMigration(brokerXml);
+
+ File output = new File("target/out.xml");
+ tool.convertQueuesToAddresses();
+
+ Properties properties = new Properties();
+ properties.put(OutputKeys.INDENT, "yes");
+ properties.put("{http://xml.apache.org/xslt}indent-amount", "3");
+ properties.put(OutputKeys.ENCODING, "UTF-8");
+ tool.write(output, properties);
+ }
+
+ @Test
+ public void scanAndReplaceTest() throws Exception {
+ File dir = new File(this.getClass().getClassLoader().getResource("replace").getPath());
+ XMLConfigurationMigration.scanAndTransform(dir);
+ }
+}
[03/17] activemq-artemis git commit: ARTEMIS-753 Queue Pause and
Resumed persisted
Posted by ma...@apache.org.
ARTEMIS-753 Queue Pause and Resumed persisted
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/a074f9f1
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/a074f9f1
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/a074f9f1
Branch: refs/heads/ARTEMIS-780
Commit: a074f9f1a5f0541a83dd2500427022eb6e63c8f9
Parents: 0f9efa9
Author: Clebert Suconic <cl...@apache.org>
Authored: Fri Oct 21 14:59:52 2016 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Fri Oct 21 20:26:23 2016 -0400
----------------------------------------------------------------------
.../api/core/management/QueueControl.java | 8 ++-
.../api/jms/management/JMSQueueControl.java | 6 ++
.../management/impl/JMSQueueControlImpl.java | 5 ++
.../core/management/impl/QueueControlImpl.java | 12 ++++
.../core/persistence/QueueBindingInfo.java | 7 ++
.../artemis/core/persistence/QueueStatus.java | 46 ++++++++++++
.../core/persistence/StorageManager.java | 11 +++
.../journal/AbstractJournalStorageManager.java | 66 ++++++++++++++++-
.../impl/journal/DescribeJournal.java | 4 ++
.../impl/journal/JournalRecordIds.java | 2 +
.../codec/PersistentQueueBindingEncoding.java | 18 +++++
.../impl/journal/codec/QueueEncoding.java | 3 +-
.../impl/journal/codec/QueueStatusEncoding.java | 75 ++++++++++++++++++++
.../impl/nullpm/NullStorageManager.java | 11 +++
.../activemq/artemis/core/server/Queue.java | 15 ++++
.../server/impl/PostOfficeJournalLoader.java | 13 +++-
.../artemis/core/server/impl/QueueImpl.java | 41 +++++++++++
.../impl/ScheduledDeliveryHandlerTest.java | 13 ++++
.../transaction/impl/TransactionImplTest.java | 11 +++
.../management/JMSQueueControlUsingJMSTest.java | 5 ++
.../management/QueueControlUsingCoreTest.java | 5 ++
.../server/QueuePeristPauseTest.java | 60 ++++++++++++++++
.../unit/core/postoffice/impl/FakeQueue.java | 15 ++++
23 files changed, 448 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/QueueControl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/QueueControl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/QueueControl.java
index 0a74d1c..3336aae 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/QueueControl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/QueueControl.java
@@ -411,9 +411,15 @@ public interface QueueControl {
void pause() throws Exception;
/**
+ * Pauses the queue. Messages are no longer delivered to its consumers.
+ */
+ @Operation(desc = "Pauses the Queue", impact = MBeanOperationInfo.ACTION)
+ void pause(@Parameter(name = "persist", desc = "if true, the pause state will be persisted.") boolean persist) throws Exception;
+
+ /**
* Resumes the queue. Messages are again delivered to its consumers.
*/
- @Operation(desc = "Resumes delivery of queued messages and gets the queue out of paused state.", impact = MBeanOperationInfo.ACTION)
+ @Operation(desc = "Resumes delivery of queued messages and gets the queue out of paused state. It will also affected the state of a persisted pause.", impact = MBeanOperationInfo.ACTION)
void resume() throws Exception;
@Operation(desc = "List all the existent consumers on the Queue")
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-jms-client/src/main/java/org/apache/activemq/artemis/api/jms/management/JMSQueueControl.java
----------------------------------------------------------------------
diff --git a/artemis-jms-client/src/main/java/org/apache/activemq/artemis/api/jms/management/JMSQueueControl.java b/artemis-jms-client/src/main/java/org/apache/activemq/artemis/api/jms/management/JMSQueueControl.java
index 56a127c..3a4101a 100644
--- a/artemis-jms-client/src/main/java/org/apache/activemq/artemis/api/jms/management/JMSQueueControl.java
+++ b/artemis-jms-client/src/main/java/org/apache/activemq/artemis/api/jms/management/JMSQueueControl.java
@@ -370,6 +370,12 @@ public interface JMSQueueControl extends DestinationControl {
void pause() throws Exception;
/**
+ * Pauses the queue. Messages are no longer delivered to its consumers.
+ */
+ @Operation(desc = "Pauses the Queue", impact = MBeanOperationInfo.ACTION)
+ void pause(@Parameter(name = "persist", desc = "if true, the pause state will be persisted.") boolean persist) throws Exception;
+
+ /**
* Returns whether the queue is paused.
*/
@Attribute(desc = "Returns true if the queue is paused.")
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-jms-server/src/main/java/org/apache/activemq/artemis/jms/management/impl/JMSQueueControlImpl.java
----------------------------------------------------------------------
diff --git a/artemis-jms-server/src/main/java/org/apache/activemq/artemis/jms/management/impl/JMSQueueControlImpl.java b/artemis-jms-server/src/main/java/org/apache/activemq/artemis/jms/management/impl/JMSQueueControlImpl.java
index a836146..36cba96 100644
--- a/artemis-jms-server/src/main/java/org/apache/activemq/artemis/jms/management/impl/JMSQueueControlImpl.java
+++ b/artemis-jms-server/src/main/java/org/apache/activemq/artemis/jms/management/impl/JMSQueueControlImpl.java
@@ -474,6 +474,11 @@ public class JMSQueueControlImpl extends StandardMBean implements JMSQueueContro
}
@Override
+ public void pause(boolean persist) throws Exception {
+ coreQueueControl.pause(persist);
+ }
+
+ @Override
public void resume() throws Exception {
coreQueueControl.resume();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
index 7275ea4..cfa8aa5 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
@@ -848,6 +848,18 @@ public class QueueControlImpl extends AbstractControl implements QueueControl {
}
}
+
+ @Override
+ public void pause(boolean persist) {
+ checkStarted();
+
+ clearIO();
+ try {
+ queue.pause(persist);
+ } finally {
+ blockOnIO();
+ }
+ }
@Override
public void resume() {
checkStarted();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java
index c05b86c..8c80a8a 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java
@@ -16,7 +16,10 @@
*/
package org.apache.activemq.artemis.core.persistence;
+import java.util.List;
+
import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.QueueStatusEncoding;
public interface QueueBindingInfo {
@@ -39,4 +42,8 @@ public interface QueueBindingInfo {
SimpleString getUser();
+ void addQueueStatusEncoding(QueueStatusEncoding status);
+
+ List<QueueStatusEncoding> getQueueStatusEncodings();
+
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueStatus.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueStatus.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueStatus.java
new file mode 100644
index 0000000..18fa9b9
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueStatus.java
@@ -0,0 +1,46 @@
+/*
+ * 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.core.persistence;
+
+public enum QueueStatus {
+ PAUSED((short) 0), RUNNING((short) 1);
+
+ public final short id;
+
+ QueueStatus(short id) {
+ this.id = id;
+ }
+
+ public static QueueStatus[] values;
+
+ static {
+ QueueStatus[] allValues = QueueStatus.values();
+ values = new QueueStatus[allValues.length];
+ for (QueueStatus v : allValues) {
+ values[v.id] = v;
+ }
+ }
+
+ public static QueueStatus fromID(short id) {
+ if (id < 0 || id > values.length) {
+ return null;
+ } else {
+ return values[id];
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
index e820664..bbfec14 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
@@ -287,6 +287,17 @@ public interface StorageManager extends IDGenerator, ActiveMQComponent {
void deleteQueueBinding(long tx, long queueBindingID) throws Exception;
+ /**
+ *
+ * @param queueID The id of the queue
+ * @param status The current status of the queue. (Reserved for future use, ATM we only use this record for PAUSED)
+ * @return the id of the journal
+ * @throws Exception
+ */
+ long storeQueueStatus(long queueID, QueueStatus status) throws Exception;
+
+ void deleteQueueStatus(long recordID) throws Exception;
+
JournalLoadInformation loadBindingJournal(List<QueueBindingInfo> queueBindingInfos,
List<GroupingInfo> groupingInfos) throws Exception;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
index cd86191..a6938d6 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
@@ -63,6 +63,7 @@ import org.apache.activemq.artemis.core.paging.impl.PageTransactionInfoImpl;
import org.apache.activemq.artemis.core.persistence.GroupingInfo;
import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.QueueBindingInfo;
+import org.apache.activemq.artemis.core.persistence.QueueStatus;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.config.PersistedAddressSetting;
import org.apache.activemq.artemis.core.persistence.config.PersistedRoles;
@@ -81,6 +82,7 @@ import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PageCount
import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PageUpdateTXEncoding;
import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PendingLargeMessageEncoding;
import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PersistentQueueBindingEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.QueueStatusEncoding;
import org.apache.activemq.artemis.core.persistence.impl.journal.codec.RefEncoding;
import org.apache.activemq.artemis.core.persistence.impl.journal.codec.ScheduledDeliveryEncoding;
import org.apache.activemq.artemis.core.persistence.impl.journal.codec.XidEncoding;
@@ -105,6 +107,7 @@ import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
import org.apache.activemq.artemis.utils.Base64;
import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.apache.activemq.artemis.utils.IDGenerator;
+import org.jboss.logging.Logger;
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ACKNOWLEDGE_CURSOR;
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_LARGE_MESSAGE_PENDING;
@@ -122,6 +125,8 @@ import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalR
*/
public abstract class AbstractJournalStorageManager implements StorageManager {
+ private static final Logger logger = Logger.getLogger(AbstractJournalStorageManager.class);
+
public enum JournalContent {
BINDINGS((byte) 0), MESSAGES((byte) 1);
@@ -1237,6 +1242,32 @@ public abstract class AbstractJournalStorageManager implements StorageManager {
}
@Override
+ public long storeQueueStatus(long queueID, QueueStatus status) throws Exception {
+ long recordID = idGenerator.generateID();
+
+ readLock();
+ try {
+ bindingsJournal.appendAddRecord(recordID, JournalRecordIds.QUEUE_STATUS_RECORD, new QueueStatusEncoding(queueID, status), true);
+ } finally {
+ readUnLock();
+ }
+
+
+ return recordID;
+ }
+
+ @Override
+ public void deleteQueueStatus(long recordID) throws Exception {
+ readLock();
+ try {
+ bindingsJournal.appendDeleteRecord(recordID, true);
+ } finally {
+ readUnLock();
+ }
+
+ }
+
+ @Override
public long storePageCounterInc(long txID, long queueID, int value) throws Exception {
readLock();
try {
@@ -1326,6 +1357,8 @@ public abstract class AbstractJournalStorageManager implements StorageManager {
JournalLoadInformation bindingsInfo = bindingsJournal.load(records, preparedTransactions, null);
+ HashMap<Long, PersistentQueueBindingEncoding> mapBindings = new HashMap<>();
+
for (RecordInfo record : records) {
long id = record.id;
@@ -1337,6 +1370,7 @@ public abstract class AbstractJournalStorageManager implements StorageManager {
PersistentQueueBindingEncoding bindingEncoding = newBindingEncoding(id, buffer);
queueBindingInfos.add(bindingEncoding);
+ mapBindings.put(bindingEncoding.getId(), bindingEncoding);
} else if (rec == JournalRecordIds.ID_COUNTER_RECORD) {
idGenerator.loadState(record.id, buffer);
} else if (rec == JournalRecordIds.GROUP_RECORD) {
@@ -1348,11 +1382,24 @@ public abstract class AbstractJournalStorageManager implements StorageManager {
} else if (rec == JournalRecordIds.SECURITY_RECORD) {
PersistedRoles roles = newSecurityRecord(id, buffer);
mapPersistedRoles.put(roles.getAddressMatch(), roles);
+ } else if (rec == JournalRecordIds.QUEUE_STATUS_RECORD) {
+ QueueStatusEncoding statusEncoding = newQueueStatusEncoding(id, buffer);
+ PersistentQueueBindingEncoding queueBindingEncoding = mapBindings.get(statusEncoding.queueID);
+ if (queueBindingEncoding != null) {
+ queueBindingEncoding.addQueueStatusEncoding(statusEncoding);
+ } else {
+ // unlikely to happen, so I didn't bother about the Logger method
+ logger.info("There is no queue with ID " + statusEncoding.queueID + ", deleting record " + statusEncoding.getId());
+ this.deleteQueueStatus(statusEncoding.getId());
+ }
} else {
- throw new IllegalStateException("Invalid record type " + rec);
+ // unlikely to happen
+ logger.warn("Invalid record type " + rec, new Exception("invalid record type " + rec));
}
}
+ mapBindings.clear(); // just to give a hand to GC
+
// This will instruct the IDGenerator to beforeStop old records
idGenerator.cleanup();
@@ -1821,6 +1868,23 @@ public abstract class AbstractJournalStorageManager implements StorageManager {
return bindingEncoding;
}
+ /**
+ * @param id
+ * @param buffer
+ * @return
+ */
+ protected static QueueStatusEncoding newQueueStatusEncoding(long id, ActiveMQBuffer buffer) {
+ QueueStatusEncoding statusEncoding = new QueueStatusEncoding();
+
+ statusEncoding.decode(buffer);
+ statusEncoding.setId(id);
+
+ return statusEncoding;
+ }
+
+
+
+
@Override
public boolean addToPage(PagingStore store,
ServerMessage msg,
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java
index 34a47bf..58723c6 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java
@@ -74,6 +74,7 @@ import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalR
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_PENDING_COUNTER;
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_TRANSACTION;
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.QUEUE_BINDING_RECORD;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.QUEUE_STATUS_RECORD;
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.SECURITY_RECORD;
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME;
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.UPDATE_DELIVERY_COUNT;
@@ -550,6 +551,9 @@ public final class DescribeJournal {
return encoding;
}
+ case QUEUE_STATUS_RECORD:
+ return AbstractJournalStorageManager.newQueueStatusEncoding(id, buffer);
+
case QUEUE_BINDING_RECORD:
return AbstractJournalStorageManager.newBindingEncoding(id, buffer);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
index 4aa470b..0169f38 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
@@ -33,6 +33,8 @@ public final class JournalRecordIds {
public static final byte QUEUE_BINDING_RECORD = 21;
+ public static final byte QUEUE_STATUS_RECORD = 22;
+
/**
* Records storing the current recordID number.
*
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java
index 4efe292..039460c 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java
@@ -16,6 +16,9 @@
*/
package org.apache.activemq.artemis.core.persistence.impl.journal.codec;
+import java.util.LinkedList;
+import java.util.List;
+
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.journal.EncodingSupport;
@@ -36,6 +39,8 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin
public SimpleString user;
+ public List<QueueStatusEncoding> queueStatusEncodings;
+
public PersistentQueueBindingEncoding() {
}
@@ -107,6 +112,19 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin
}
@Override
+ public void addQueueStatusEncoding(QueueStatusEncoding status) {
+ if (queueStatusEncodings == null) {
+ queueStatusEncodings = new LinkedList<>();
+ }
+ queueStatusEncodings.add(status);
+ }
+
+ @Override
+ public List<QueueStatusEncoding> getQueueStatusEncodings() {
+ return queueStatusEncodings;
+ }
+
+ @Override
public void decode(final ActiveMQBuffer buffer) {
name = buffer.readSimpleString();
address = buffer.readSimpleString();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/QueueEncoding.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/QueueEncoding.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/QueueEncoding.java
index 1e05195..9220509 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/QueueEncoding.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/QueueEncoding.java
@@ -18,6 +18,7 @@ package org.apache.activemq.artemis.core.persistence.impl.journal.codec;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.core.journal.EncodingSupport;
+import org.apache.activemq.artemis.utils.DataConstants;
public class QueueEncoding implements EncodingSupport {
@@ -44,7 +45,7 @@ public class QueueEncoding implements EncodingSupport {
@Override
public int getEncodeSize() {
- return 8;
+ return DataConstants.SIZE_LONG;
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/QueueStatusEncoding.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/QueueStatusEncoding.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/QueueStatusEncoding.java
new file mode 100644
index 0000000..fe2b1f5
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/QueueStatusEncoding.java
@@ -0,0 +1,75 @@
+/**
+ * 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.core.persistence.impl.journal.codec;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.core.persistence.QueueStatus;
+import org.apache.activemq.artemis.utils.DataConstants;
+
+public class QueueStatusEncoding extends QueueEncoding {
+
+ private QueueStatus status;
+
+ private long id;
+
+ public QueueStatusEncoding(long queueID, QueueStatus status) {
+ super(queueID);
+ this.status = status;
+ }
+
+ public QueueStatusEncoding() {
+ super();
+ }
+
+ @Override
+ public void decode(final ActiveMQBuffer buffer) {
+ super.decode(buffer);
+ short shortStatus = buffer.readShort();
+ this.status = QueueStatus.fromID(shortStatus);
+ }
+
+ @Override
+ public void encode(final ActiveMQBuffer buffer) {
+ super.encode(buffer);
+ buffer.writeShort(status.id);
+ }
+
+ public QueueStatus getStatus() {
+ return status;
+ }
+
+ public long getId() {
+ return id;
+ }
+
+ public QueueStatusEncoding setId(long id) {
+ this.id = id;
+ return this;
+ }
+
+ @Override
+ public int getEncodeSize() {
+ return super.getEncodeSize() + DataConstants.SIZE_SHORT;
+ }
+
+ @Override
+ public String toString() {
+ return "QueueStatusEncoding [id=" + id + ", queueID=" + queueID + ", status=" + status + "]";
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
index f13d2fa..3a2999e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
@@ -41,6 +41,7 @@ import org.apache.activemq.artemis.core.paging.cursor.PagePosition;
import org.apache.activemq.artemis.core.persistence.GroupingInfo;
import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.QueueBindingInfo;
+import org.apache.activemq.artemis.core.persistence.QueueStatus;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.config.PersistedAddressSetting;
import org.apache.activemq.artemis.core.persistence.config.PersistedRoles;
@@ -85,6 +86,16 @@ public class NullStorageManager implements StorageManager {
}
@Override
+ public long storeQueueStatus(long queueID, QueueStatus status) throws Exception {
+ return 0;
+ }
+
+ @Override
+ public void deleteQueueStatus(long recordID) throws Exception {
+
+ }
+
+ @Override
public void injectMonitor(FileStoreMonitor monitor) throws Exception {
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
index a9a87c3..0dcef3d 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
@@ -207,6 +207,15 @@ public interface Queue extends Bindable {
void pause();
/**
+ * Pauses the queue. It will receive messages but won't give them to the consumers until resumed.
+ * If a queue is paused, pausing it again will only throw a warning.
+ * To check if a queue is paused, invoke <i>isPaused()</i>
+ */
+ void pause(boolean persist);
+
+ void reloadPause(long recordID);
+
+ /**
* Resumes the delivery of message for the queue.
* If a queue is resumed, resuming it again will only throw a warning.
* To check if a queue is resumed, invoke <i>isPaused()</i>
@@ -218,6 +227,12 @@ public interface Queue extends Bindable {
*/
boolean isPaused();
+ /**
+ * if the pause was persisted
+ * @return
+ */
+ boolean isPersistedPause();
+
Executor getExecutor();
void resetAllIterators();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
index ccb00cb..9a8ae74 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
@@ -39,9 +39,11 @@ import org.apache.activemq.artemis.core.paging.cursor.PageSubscriptionCounter;
import org.apache.activemq.artemis.core.paging.impl.Page;
import org.apache.activemq.artemis.core.persistence.GroupingInfo;
import org.apache.activemq.artemis.core.persistence.QueueBindingInfo;
+import org.apache.activemq.artemis.core.persistence.QueueStatus;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.impl.PageCountPending;
import org.apache.activemq.artemis.core.persistence.impl.journal.AddMessageRecord;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.QueueStatusEncoding;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.DuplicateIDCache;
import org.apache.activemq.artemis.core.postoffice.PostOffice;
@@ -146,6 +148,13 @@ public class PostOfficeJournalLoader implements JournalLoader {
queue.setConsumersRefCount(new AutoCreatedQueueManagerImpl(((PostOfficeImpl) postOffice).getServer().getJMSQueueDeleter(), queueBindingInfo.getQueueName()));
}
+ if (queueBindingInfo.getQueueStatusEncodings() != null) {
+ for (QueueStatusEncoding encoding : queueBindingInfo.getQueueStatusEncodings()) {
+ if (encoding.getStatus() == QueueStatus.PAUSED)
+ queue.reloadPause(encoding.getId());
+ }
+ }
+
final Binding binding = new LocalQueueBinding(queue.getAddress(), queue, nodeManager.getNodeId());
queues.put(queue.getID(), queue);
postOffice.addBinding(binding);
@@ -245,7 +254,9 @@ public class PostOfficeJournalLoader implements JournalLoader {
ResourceManager resourceManager,
Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap) throws Exception {
for (Queue queue : queues.values()) {
- queue.resume();
+ if (!queue.isPersistedPause()) {
+ queue.resume();
+ }
}
if (System.getProperty("org.apache.activemq.opt.directblast") != null) {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index 7c8ad0a..d30544f 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -51,6 +51,7 @@ import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.message.impl.MessageImpl;
import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
+import org.apache.activemq.artemis.core.persistence.QueueStatus;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.Bindings;
@@ -176,6 +177,8 @@ public class QueueImpl implements Queue {
private boolean paused;
+ private long pauseStatusRecord = -1;
+
private static final int MAX_SCHEDULED_RUNNERS = 2;
// We don't ever need more than two DeliverRunner on the executor's list
@@ -1718,8 +1721,32 @@ public class QueueImpl implements Queue {
@Override
public synchronized void pause() {
+ pause(false);
+ }
+
+ @Override
+ public synchronized void reloadPause(long recordID) {
+ this.paused = true;
+ if (pauseStatusRecord >= 0) {
+ try {
+ storageManager.deleteQueueStatus(pauseStatusRecord);
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ }
+ }
+ this.pauseStatusRecord = recordID;
+ }
+
+ @Override
+ public synchronized void pause(boolean persist) {
try {
this.flushDeliveriesInTransit();
+ if (persist && isDurable()) {
+ if (pauseStatusRecord >= 0) {
+ storageManager.deleteQueueStatus(pauseStatusRecord);
+ }
+ pauseStatusRecord = storageManager.storeQueueStatus(this.id, QueueStatus.PAUSED);
+ }
} catch (Exception e) {
ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
}
@@ -1730,6 +1757,15 @@ public class QueueImpl implements Queue {
public synchronized void resume() {
paused = false;
+ if (pauseStatusRecord >= 0) {
+ try {
+ storageManager.deleteQueueStatus(pauseStatusRecord);
+ } catch (Exception e) {
+ ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
+ }
+ pauseStatusRecord = -1;
+ }
+
deliverAsync();
}
@@ -1739,6 +1775,11 @@ public class QueueImpl implements Queue {
}
@Override
+ public synchronized boolean isPersistedPause() {
+ return this.pauseStatusRecord >= 0;
+ }
+
+ @Override
public boolean isDirectDeliver() {
return directDeliver;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
index 91e5e7a..d82f7d3 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
@@ -845,6 +845,11 @@ public class ScheduledDeliveryHandlerTest extends Assert {
this.expectedElements = new CountDownLatch(expectedElements);
}
+ @Override
+ public boolean isPersistedPause() {
+ return false;
+ }
+
public boolean waitCompletion(long timeout, TimeUnit timeUnit) throws Exception {
return expectedElements.await(timeout, timeUnit);
}
@@ -863,6 +868,14 @@ public class ScheduledDeliveryHandlerTest extends Assert {
}
@Override
+ public void pause(boolean persist) {
+ }
+
+ @Override
+ public void reloadPause(long recordID) {
+ }
+
+ @Override
public Filter getFilter() {
return null;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
index 0c2fec5..93c5c9d 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
@@ -41,6 +41,7 @@ import org.apache.activemq.artemis.core.paging.cursor.PagePosition;
import org.apache.activemq.artemis.core.persistence.GroupingInfo;
import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.QueueBindingInfo;
+import org.apache.activemq.artemis.core.persistence.QueueStatus;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.config.PersistedAddressSetting;
import org.apache.activemq.artemis.core.persistence.config.PersistedRoles;
@@ -243,6 +244,16 @@ public class TransactionImplTest extends ActiveMQTestBase {
}
@Override
+ public long storeQueueStatus(long queueID, QueueStatus status) throws Exception {
+ return 0;
+ }
+
+ @Override
+ public void deleteQueueStatus(long recordID) throws Exception {
+
+ }
+
+ @Override
public void pageWrite(PagedMessage message, int pageNumber) {
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java
index f6c4bde..7a07439 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java
@@ -351,6 +351,11 @@ public class JMSQueueControlUsingJMSTest extends JMSQueueControlTest {
}
@Override
+ public void pause(boolean persist) throws Exception {
+ proxy.invokeOperation("pause", persist);
+ }
+
+ @Override
public void resume() throws Exception {
proxy.invokeOperation("resume");
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlUsingCoreTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlUsingCoreTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlUsingCoreTest.java
index e8e04f2..9b901fc 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlUsingCoreTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlUsingCoreTest.java
@@ -337,6 +337,11 @@ public class QueueControlUsingCoreTest extends QueueControlTest {
}
@Override
+ public void pause(boolean persist) throws Exception {
+ proxy.invokeOperation("pause", persist);
+ }
+
+ @Override
public void resume() throws Exception {
proxy.invokeOperation("resume");
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/QueuePeristPauseTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/QueuePeristPauseTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/QueuePeristPauseTest.java
new file mode 100644
index 0000000..e383c85
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/QueuePeristPauseTest.java
@@ -0,0 +1,60 @@
+/**
+ * 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.server;
+
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class QueuePeristPauseTest extends ActiveMQTestBase {
+
+ @Test
+ public void testPauseQueue() throws Exception {
+ ActiveMQServer server = createServer(true, false);
+ server.start();
+
+ Queue queue = server.createQueue(SimpleString.toSimpleString("q1"),
+ SimpleString.toSimpleString("q1"),
+ null, true, false);
+
+ queue.pause(true);
+
+ server.stop();
+ server.start();
+
+ for (int i = 0; i < 4; i++) {
+ server.stop();
+ server.start();
+ queue = server.locateQueue(SimpleString.toSimpleString("q1"));
+ Assert.assertTrue(queue.isPaused());
+ }
+
+ queue.resume();
+
+ for (int i = 0; i < 4; i++) {
+ server.stop();
+ server.start();
+ queue = server.locateQueue(SimpleString.toSimpleString("q1"));
+ Assert.assertFalse(queue.isPaused());
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a074f9f1/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
----------------------------------------------------------------------
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
index 8eae7d6..bba5dc1 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
@@ -57,6 +57,16 @@ public class FakeQueue implements Queue {
}
@Override
+ public void reloadPause(long recordID) {
+
+ }
+
+ @Override
+ public boolean isPersistedPause() {
+ return false;
+ }
+
+ @Override
public int retryMessages(Filter filter) throws Exception {
return 0;
}
@@ -103,6 +113,11 @@ public class FakeQueue implements Queue {
}
@Override
+ public void pause(boolean persist) {
+
+ }
+
+ @Override
public boolean flushExecutor() {
return true;
}
[07/17] activemq-artemis git commit: Upgrade Jolokia version
Posted by ma...@apache.org.
Upgrade Jolokia version
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/e4d58ce5
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/e4d58ce5
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/e4d58ce5
Branch: refs/heads/ARTEMIS-780
Commit: e4d58ce59676482a018a0f591877e191f7aebd4f
Parents: e5df63c
Author: Paul Gallagher <pg...@redhat.com>
Authored: Mon Oct 24 12:28:49 2016 +0100
Committer: Paul Gallagher <pg...@redhat.com>
Committed: Mon Oct 24 12:28:49 2016 +0100
----------------------------------------------------------------------
.../activemq/artemis/cli/commands/etc/bootstrap-web-settings.txt | 2 +-
docs/user-manual/en/security.md | 4 ++--
pom.xml | 2 +-
3 files changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e4d58ce5/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/bootstrap-web-settings.txt
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/bootstrap-web-settings.txt b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/bootstrap-web-settings.txt
index 49c5e37..946b485 100644
--- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/bootstrap-web-settings.txt
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/bootstrap-web-settings.txt
@@ -1,4 +1,4 @@
<!-- The web server is only bound to loalhost by default -->
<web bind="${web.protocol}://localhost:${http.port}" path="web"${extra.web.attributes}>
- <app url="jolokia" war="jolokia-war-1.3.3.war"/>
+ <app url="jolokia" war="jolokia-war-1.3.5.war"/>
</web>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e4d58ce5/docs/user-manual/en/security.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/security.md b/docs/user-manual/en/security.md
index dfa9f46..2961cbd 100644
--- a/docs/user-manual/en/security.md
+++ b/docs/user-manual/en/security.md
@@ -666,7 +666,7 @@ Artemis comes with a web console that allows user to browse Artemis documentatio
web access is plain HTTP. It is configured in `bootstrap.xml`:
<web bind="http://localhost:8161" path="web">
- <app url="jolokia" war="jolokia-war-1.3.3.war"/>
+ <app url="jolokia" war="jolokia-war-1.3.5.war"/>
</web>
Alternatively you can edit the above configuration to enable secure access using HTTPS protocol. e.g.:
@@ -675,7 +675,7 @@ Alternatively you can edit the above configuration to enable secure access using
path="web"
keyStorePath="${artemis.instance}/etc/keystore.jks"
keyStorePassword="password">
- <app url="jolokia" war="jolokia-war-1.3.3.war"/>
+ <app url="jolokia" war="jolokia-war-1.3.5.war"/>
</web>
As shown in the example, to enable https the first thing to do is config the `bind` to be an `https` url. In addition,
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e4d58ce5/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 5a05c4f..d040375 100644
--- a/pom.xml
+++ b/pom.xml
@@ -558,7 +558,7 @@
<groupId>org.jolokia</groupId>
<artifactId>jolokia-war</artifactId>
<type>war</type>
- <version>1.3.3</version>
+ <version>1.3.5</version>
<!-- License: Apache 2.0 -->
</dependency>
<!-- ## End Jetty Wed Dependencies ## -->
[06/17] activemq-artemis git commit: This closes #858
Posted by ma...@apache.org.
This closes #858
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/e5df63c2
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/e5df63c2
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/e5df63c2
Branch: refs/heads/ARTEMIS-780
Commit: e5df63c24156730d4a4beaa3ced04a78aeacedd0
Parents: 8d4f507 e95e4f7
Author: Clebert Suconic <cl...@apache.org>
Authored: Sat Oct 22 10:41:21 2016 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Sat Oct 22 10:41:21 2016 -0400
----------------------------------------------------------------------
.../artemis/core/server/ActiveMQServerLogger.java | 10 ++++++++++
.../activemq/artemis/core/server/impl/QueueImpl.java | 10 ++++++++++
2 files changed, 20 insertions(+)
----------------------------------------------------------------------
[08/17] activemq-artemis git commit: This closes #859
Posted by ma...@apache.org.
This closes #859
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/5315acb8
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/5315acb8
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/5315acb8
Branch: refs/heads/ARTEMIS-780
Commit: 5315acb8fb8eedd26612ff8a5bfb8afad0c8264c
Parents: e5df63c e4d58ce
Author: Clebert Suconic <cl...@apache.org>
Authored: Mon Oct 24 08:49:08 2016 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Mon Oct 24 08:49:08 2016 -0400
----------------------------------------------------------------------
.../activemq/artemis/cli/commands/etc/bootstrap-web-settings.txt | 2 +-
docs/user-manual/en/security.md | 4 ++--
pom.xml | 2 +-
3 files changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
[09/17] activemq-artemis git commit: ARTEMIS-782 Added configuration
elements for new address model
Posted by ma...@apache.org.
ARTEMIS-782 Added configuration elements for new address model
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/8c51ed73
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/8c51ed73
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/8c51ed73
Branch: refs/heads/ARTEMIS-780
Commit: 8c51ed739699460e1add2cbfae343e64c3b5c0c6
Parents: 5315acb
Author: Martyn Taylor <mt...@redhat.com>
Authored: Tue Oct 18 19:45:02 2016 +0100
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Mon Oct 24 14:33:52 2016 +0100
----------------------------------------------------------------------
.../config/ActiveMQDefaultConfiguration.java | 12 ++
.../artemis/core/config/Configuration.java | 15 ++
.../core/config/CoreAddressConfiguration.java | 145 +++++++++++++++++++
.../core/config/CoreQueueConfiguration.java | 43 ++++++
.../core/config/impl/ConfigurationImpl.java | 20 +++
.../deployers/impl/FileConfigurationParser.java | 80 +++++++++-
.../resources/schema/artemis-configuration.xsd | 78 +++++++++-
.../impl/DefaultsFileConfigurationTest.java | 2 +
.../core/config/impl/FileConfigurationTest.java | 63 ++++++++
.../resources/ConfigurationTest-full-config.xml | 26 ++++
10 files changed, 478 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8c51ed73/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 e07493f..04d06c0 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
@@ -435,6 +435,10 @@ public final class ActiveMQDefaultConfiguration {
public static final int DEFAULT_DISK_SCAN = 5000;
+ public static final int DEFAULT_MAX_QUEUE_CONSUMERS = -1;
+
+ public static final boolean DEFAULT_DELETE_QUEUE_ON_NO_CONSUMERS = false;
+
/**
* 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.
*/
@@ -1168,4 +1172,12 @@ public final class ActiveMQDefaultConfiguration {
public static int getDefaultDiskScanPeriod() {
return DEFAULT_DISK_SCAN;
}
+
+ public static int getDefaultMaxQueueConsumers() {
+ return DEFAULT_MAX_QUEUE_CONSUMERS;
+ }
+
+ public static boolean getDefaultDeleteQueueOnNoConsumers() {
+ return DEFAULT_DELETE_QUEUE_ON_NO_CONSUMERS;
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8c51ed73/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 f486a88..17a305e 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
@@ -387,6 +387,21 @@ public interface Configuration {
Configuration addQueueConfiguration(final CoreQueueConfiguration config);
/**
+ * Returns the addresses configured for this server.
+ */
+ List<CoreAddressConfiguration> getAddressConfigurations();
+
+ /**
+ * Sets the addresses configured for this server.
+ */
+ Configuration setAddressConfigurations(final List<CoreAddressConfiguration> configs);
+
+ /**
+ * Adds an addresses configuration
+ */
+ Configuration addAddressConfiguration(final CoreAddressConfiguration config);
+
+ /**
* Returns the management address of this server. <br>
* Clients can send management messages to this address to manage this server. <br>
* Default value is {@link org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration#DEFAULT_MANAGEMENT_ADDRESS}.
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8c51ed73/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
new file mode 100644
index 0000000..cb6d43f
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreAddressConfiguration.java
@@ -0,0 +1,145 @@
+/*
+ * 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.core.config;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
+
+public class CoreAddressConfiguration implements Serializable {
+
+ public enum RoutingType {
+ MULTICAST,
+ ANYCAST
+ }
+
+ private String name = null;
+
+ private RoutingType routingType = null;
+
+ private Integer defaultMaxConsumers = ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers();
+
+ private Boolean defaultDeleteOnNoConsumers = ActiveMQDefaultConfiguration.getDefaultDeleteQueueOnNoConsumers();
+
+ private List<CoreQueueConfiguration> queueConfigurations = new ArrayList<>();
+
+ public CoreAddressConfiguration() {
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public CoreAddressConfiguration setName(String name) {
+ this.name = name;
+ return this;
+ }
+
+ public RoutingType getRoutingType() {
+ return routingType;
+ }
+
+ public CoreAddressConfiguration setRoutingType(RoutingType routingType) {
+ this.routingType = routingType;
+ return this;
+ }
+
+ public CoreAddressConfiguration setQueueConfigurations(List<CoreQueueConfiguration> queueConfigurations) {
+ this.queueConfigurations = queueConfigurations;
+ return this;
+ }
+
+ public CoreAddressConfiguration addQueueConfiguration(CoreQueueConfiguration queueConfiguration) {
+ this.queueConfigurations.add(queueConfiguration);
+ return this;
+ }
+
+ public List<CoreQueueConfiguration> getQueueConfigurations() {
+ return queueConfigurations;
+ }
+
+ public Boolean getDefaultDeleteOnNoConsumers() {
+ return defaultDeleteOnNoConsumers;
+ }
+
+ public CoreAddressConfiguration setDefaultDeleteOnNoConsumers(Boolean defaultDeleteOnNoConsumers) {
+ this.defaultDeleteOnNoConsumers = defaultDeleteOnNoConsumers;
+ return this;
+ }
+
+ public Integer getDefaultMaxConsumers() {
+ return defaultMaxConsumers;
+ }
+
+ public CoreAddressConfiguration setDefaultMaxConsumers(Integer defaultMaxConsumers) {
+ this.defaultMaxConsumers = defaultMaxConsumers;
+ return this;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((name == null) ? 0 : name.hashCode());
+ result = prime * result + ((routingType == null) ? 0 : routingType.hashCode());
+ result = prime * result + ((queueConfigurations == null) ? 0 : queueConfigurations.hashCode());
+ result = prime * result + ((defaultMaxConsumers == null) ? 0 : defaultMaxConsumers.hashCode());
+ result = prime * result + ((defaultDeleteOnNoConsumers == null) ? 0 : defaultDeleteOnNoConsumers.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ CoreAddressConfiguration other = (CoreAddressConfiguration) obj;
+ if (name == null) {
+ if (other.name != null)
+ return false;
+ } else if (!name.equals(other.name))
+ return false;
+ if (routingType == null) {
+ if (other.routingType != null)
+ return false;
+ } else if (!routingType.equals(other.routingType))
+ return false;
+ if (queueConfigurations == null) {
+ if (other.queueConfigurations != null)
+ return false;
+ } else if (!queueConfigurations.equals(other.queueConfigurations))
+ return false;
+ if (defaultMaxConsumers == null) {
+ if (other.defaultMaxConsumers != null)
+ return false;
+ } else if (!defaultMaxConsumers.equals(other.defaultMaxConsumers))
+ return false;
+ if (defaultDeleteOnNoConsumers == null) {
+ if (other.defaultDeleteOnNoConsumers != null)
+ return false;
+ } else if (!defaultDeleteOnNoConsumers.equals(other.defaultDeleteOnNoConsumers)) {
+ return false;
+ }
+
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8c51ed73/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreQueueConfiguration.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreQueueConfiguration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreQueueConfiguration.java
index 2e7b9ca..79b2fd2 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreQueueConfiguration.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreQueueConfiguration.java
@@ -30,6 +30,10 @@ public class CoreQueueConfiguration implements Serializable {
private boolean durable = true;
+ private Integer maxConsumers = null;
+
+ private Boolean deleteOnNoConsumers = null;
+
public CoreQueueConfiguration() {
}
@@ -49,6 +53,8 @@ public class CoreQueueConfiguration implements Serializable {
return durable;
}
+
+
/**
* @param address the address to set
*/
@@ -81,6 +87,30 @@ public class CoreQueueConfiguration implements Serializable {
return this;
}
+ /**
+ * @param maxConsumers for this queue, default is -1 (unlimited)
+ */
+ public CoreQueueConfiguration setMaxConsumers(Integer maxConsumers) {
+ this.maxConsumers = maxConsumers;
+ return this;
+ }
+
+ /**
+ * @param deleteOnNoConsumers delete this queue when consumer count reaches 0, default is false
+ */
+ public CoreQueueConfiguration setDeleteOnNoConsumers(Boolean deleteOnNoConsumers) {
+ this.deleteOnNoConsumers = deleteOnNoConsumers;
+ return this;
+ }
+
+ public Boolean getDeleteOnNoConsumers() {
+ return deleteOnNoConsumers;
+ }
+
+ public Integer getMaxConsumers() {
+ return maxConsumers;
+ }
+
@Override
public int hashCode() {
final int prime = 31;
@@ -89,6 +119,8 @@ public class CoreQueueConfiguration implements Serializable {
result = prime * result + (durable ? 1231 : 1237);
result = prime * result + ((filterString == null) ? 0 : filterString.hashCode());
result = prime * result + ((name == null) ? 0 : name.hashCode());
+ result = prime * result + ((maxConsumers == null) ? 0 : maxConsumers.hashCode());
+ result = prime * result + ((deleteOnNoConsumers == null) ? 0 : deleteOnNoConsumers.hashCode());
return result;
}
@@ -118,6 +150,17 @@ public class CoreQueueConfiguration implements Serializable {
return false;
} else if (!name.equals(other.name))
return false;
+ if (maxConsumers == null) {
+ if (other.maxConsumers != null)
+ return false;
+ } else if (!maxConsumers.equals(other.maxConsumers))
+ return false;
+ if (deleteOnNoConsumers == null) {
+ if (other.deleteOnNoConsumers != null)
+ return false;
+ } else if (!deleteOnNoConsumers.equals(other.deleteOnNoConsumers)) {
+ return false;
+ }
return true;
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8c51ed73/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 2613929..8ff1922 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
@@ -47,6 +47,7 @@ import org.apache.activemq.artemis.core.config.BridgeConfiguration;
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.ConnectorServiceConfiguration;
+import org.apache.activemq.artemis.core.config.CoreAddressConfiguration;
import org.apache.activemq.artemis.core.config.CoreQueueConfiguration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.config.HAPolicyConfiguration;
@@ -127,6 +128,8 @@ public class ConfigurationImpl implements Configuration, Serializable {
private List<CoreQueueConfiguration> queueConfigurations = new ArrayList<>();
+ private List<CoreAddressConfiguration> addressConfigurations = new ArrayList<>();
+
protected transient List<BroadcastGroupConfiguration> broadcastGroupConfigurations = new ArrayList<>();
protected transient Map<String, DiscoveryGroupConfiguration> discoveryGroupConfigurations = new LinkedHashMap<>();
@@ -582,6 +585,23 @@ public class ConfigurationImpl implements Configuration, Serializable {
}
@Override
+ public List<CoreAddressConfiguration> getAddressConfigurations() {
+ return addressConfigurations;
+ }
+
+ @Override
+ public Configuration setAddressConfigurations(List<CoreAddressConfiguration> configs) {
+ this.addressConfigurations = configs;
+ return this;
+ }
+
+ @Override
+ public Configuration addAddressConfiguration(CoreAddressConfiguration config) {
+ this.addressConfigurations.add(config);
+ return this;
+ }
+
+ @Override
public Map<String, DiscoveryGroupConfiguration> getDiscoveryGroupConfigurations() {
return discoveryGroupConfigurations;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8c51ed73/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 d90c343..fc045b3 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
@@ -43,6 +43,7 @@ import org.apache.activemq.artemis.core.config.BridgeConfiguration;
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.ConnectorServiceConfiguration;
+import org.apache.activemq.artemis.core.config.CoreAddressConfiguration;
import org.apache.activemq.artemis.core.config.CoreQueueConfiguration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.config.ScaleDownConfiguration;
@@ -542,6 +543,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
parseQueues(e, config);
+ parseAddresses(e, config);
+
parseSecurity(e, config);
NodeList connectorServiceConfigs = e.getElementsByTagName("connector-service");
@@ -585,13 +588,35 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
*/
private void parseQueues(final Element e, final Configuration config) {
NodeList elements = e.getElementsByTagName("queues");
+ if (elements.getLength() != 0) {
+ Element node = (Element) elements.item(0);
+ config.setQueueConfigurations(parseQueueConfigurations(node));
+ }
+ }
+
+ private List<CoreQueueConfiguration> parseQueueConfigurations(final Element node) {
+ List<CoreQueueConfiguration> queueConfigurations = new ArrayList<>();
+ NodeList list = node.getElementsByTagName("queue");
+ for (int i = 0; i < list.getLength(); i++) {
+ CoreQueueConfiguration queueConfig = parseQueueConfiguration(list.item(i));
+ queueConfigurations.add(queueConfig);
+ }
+ return queueConfigurations;
+ }
+
+ /**
+ * @param e
+ * @param config
+ */
+ private void parseAddresses(final Element e, final Configuration config) {
+ NodeList elements = e.getElementsByTagName("addresses");
if (elements.getLength() != 0) {
Element node = (Element) elements.item(0);
- NodeList list = node.getElementsByTagName("queue");
+ NodeList list = node.getElementsByTagName("address");
for (int i = 0; i < list.getLength(); i++) {
- CoreQueueConfiguration queueConfig = parseQueueConfiguration(list.item(i));
- config.getQueueConfigurations().add(queueConfig);
+ CoreAddressConfiguration addrConfig = parseAddressConfiguration(list.item(i));
+ config.getAddressConfigurations().add(addrConfig);
}
}
}
@@ -829,9 +854,20 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
String address = null;
String filterString = null;
boolean durable = true;
+ Integer maxConsumers = null;
+ Boolean deleteOnNoConsumers = null;
+
+ NamedNodeMap attributes = node.getAttributes();
+ for (int i = 0; i < attributes.getLength(); i++) {
+ Node item = attributes.item(i);
+ if (item.getNodeName().equals("max-consumers")) {
+ maxConsumers = Integer.parseInt(item.getNodeValue());
+ } else if (item.getNodeName().equals("delete-on-no-consumers")) {
+ deleteOnNoConsumers = Boolean.parseBoolean(item.getNodeValue());
+ }
+ }
NodeList children = node.getChildNodes();
-
for (int j = 0; j < children.getLength(); j++) {
Node child = children.item(j);
@@ -844,7 +880,41 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
}
}
- return new CoreQueueConfiguration().setAddress(address).setName(name).setFilterString(filterString).setDurable(durable);
+ return new CoreQueueConfiguration()
+ .setAddress(address)
+ .setName(name)
+ .setFilterString(filterString)
+ .setDurable(durable)
+ .setMaxConsumers(maxConsumers)
+ .setDeleteOnNoConsumers(deleteOnNoConsumers);
+ }
+
+ protected CoreAddressConfiguration parseAddressConfiguration(final Node node) {
+ String name = getAttributeValue(node, "name");
+ String routingType = getAttributeValue(node, "type");
+
+ CoreAddressConfiguration addressConfiguration = new CoreAddressConfiguration();
+ addressConfiguration.setName(name)
+ .setRoutingType(CoreAddressConfiguration.RoutingType.valueOf(routingType.toUpperCase()));
+
+ NodeList children = node.getChildNodes();
+ for (int j = 0; j < children.getLength(); j++) {
+ Node child = children.item(j);
+ if (child.getNodeName().equals("queues")) {
+ addressConfiguration.setQueueConfigurations(parseQueueConfigurations((Element) child));
+ }
+ }
+
+ for (CoreQueueConfiguration coreQueueConfiguration : addressConfiguration.getQueueConfigurations()) {
+ coreQueueConfiguration.setAddress(addressConfiguration.getName());
+ if (coreQueueConfiguration.getMaxConsumers() == null) {
+ coreQueueConfiguration.setMaxConsumers(addressConfiguration.getDefaultMaxConsumers());
+ }
+ if (coreQueueConfiguration.getDeleteOnNoConsumers() == null) {
+ coreQueueConfiguration.setDeleteOnNoConsumers(addressConfiguration.getDefaultDeleteOnNoConsumers());
+ }
+ }
+ return addressConfiguration;
}
private TransportConfiguration parseAcceptorTransportConfiguration(final Element e,
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8c51ed73/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 73aa20b..4c3e068 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -422,7 +422,6 @@
</xsd:complexType>
</xsd:element>
-
<!-- QUEUES -->
<xsd:element name="queues" maxOccurs="1" minOccurs="0">
<xsd:annotation>
@@ -855,6 +854,8 @@
</xsd:sequence>
</xsd:complexType>
</xsd:element>
+
+ <xsd:element name="addresses" type="addressesType" maxOccurs="1" minOccurs="0" />
</xsd:all>
</xsd:complexType>
@@ -2512,4 +2513,79 @@
</xsd:extension>
</xsd:simpleContent>
</xsd:complexType>
+
+
+ <!-- 2.0 Addressing configuration -->
+ <xsd:simpleType name="routingType">
+ <xsd:restriction base="xsd:string">
+ <xsd:enumeration value="multicast" />
+ <xsd:enumeration value="anycast" />
+ </xsd:restriction>
+ </xsd:simpleType>
+
+ <xsd:complexType name="queueType">
+ <xsd:all>
+ <xsd:element ref="filter" maxOccurs="1" minOccurs="0"/>
+ <xsd:element name="durable" type="xsd:boolean" default="true" maxOccurs="1" minOccurs="0" />
+ </xsd:all>
+ <xsd:attribute name="name" type="xsd:ID" use="required"/>
+ <xsd:attribute name="max-consumers" type="xsd:integer" use="optional"/>
+ <xsd:attribute name="delete-on-no-consumers" type="xsd:boolean" use="optional"/>
+ </xsd:complexType>
+
+ <xsd:complexType name="addressType">
+ <xsd:all>
+ <xsd:element name="queues" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ a list of pre configured queues to create
+ </xsd:documentation>
+ </xsd:annotation>
+ <xsd:complexType>
+ <xsd:sequence>
+ <xsd:element name="queue" type="queueType" maxOccurs="unbounded" minOccurs="0" />
+ </xsd:sequence>
+ </xsd:complexType>
+ </xsd:element>
+ </xsd:all>
+ <xsd:attribute name="name" type="xsd:string" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ The address name to matches incoming message addresses
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ <xsd:attribute name="type" type="routingType" use="required">
+ <xsd:annotation>
+ <xsd:documentation>
+ The address name to matches incoming message addresses
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ <xsd:attribute name="default-max-consumers" type="xsd:int" use="optional" default="-1">
+ <xsd:annotation>
+ <xsd:documentation>
+ The default value of max-consumers applied to all queues that are
+ auto-created under this address. Also applies to any queues that do not
+ specify a value for max-consumers.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ <xsd:attribute name="default-delete-on-no-consumers" type="xsd:boolean" use="optional" default="false">
+ <xsd:annotation>
+ <xsd:documentation>
+ The default value of delete-on-no-consumers applied to all queues that are
+ auto-created under this address. Also applies to any queues that do not
+ specify a value for delete-on-no-consumers.
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
+ </xsd:complexType>
+
+ <xsd:complexType name="addressesType">
+ <xsd:sequence>
+ <xsd:element name="address" type="addressType" maxOccurs="unbounded" minOccurs="0"/>
+ </xsd:sequence>
+ </xsd:complexType>
+
</xsd:schema>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8c51ed73/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/DefaultsFileConfigurationTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/DefaultsFileConfigurationTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/DefaultsFileConfigurationTest.java
index 700c290..07d5f58 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/DefaultsFileConfigurationTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/DefaultsFileConfigurationTest.java
@@ -64,6 +64,8 @@ public class DefaultsFileConfigurationTest extends ConfigurationImplTest {
Assert.assertEquals(Collections.emptyList(), conf.getQueueConfigurations());
+ Assert.assertEquals(Collections.emptyList(), conf.getAddressConfigurations());
+
Assert.assertEquals(ActiveMQDefaultConfiguration.getDefaultManagementAddress(), conf.getManagementAddress());
Assert.assertEquals(ActiveMQDefaultConfiguration.getDefaultManagementNotificationAddress(), conf.getManagementNotificationAddress());
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8c51ed73/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
index a2afd97..d004a22 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
@@ -35,6 +35,8 @@ import org.apache.activemq.artemis.api.core.UDPBroadcastEndpointFactory;
import org.apache.activemq.artemis.core.config.BridgeConfiguration;
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.config.CoreAddressConfiguration;
+import org.apache.activemq.artemis.core.config.CoreQueueConfiguration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.config.FileDeploymentManager;
import org.apache.activemq.artemis.core.config.HAPolicyConfiguration;
@@ -49,6 +51,9 @@ import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.junit.Assert;
import org.junit.Test;
+import static org.apache.activemq.artemis.core.config.CoreAddressConfiguration.RoutingType.ANYCAST;
+import static org.apache.activemq.artemis.core.config.CoreAddressConfiguration.RoutingType.MULTICAST;
+
public class FileConfigurationTest extends ConfigurationImplTest {
private final String fullConfigurationName = "ConfigurationTest-full-config.xml";
@@ -324,6 +329,8 @@ public class FileConfigurationTest extends ConfigurationImplTest {
assertEquals("color='blue'", conf.getQueueConfigurations().get(1).getFilterString());
assertEquals(false, conf.getQueueConfigurations().get(1).isDurable());
+ verifyAddresses();
+
Map<String, Set<Role>> roles = conf.getSecurityRoles();
assertEquals(2, roles.size());
@@ -356,6 +363,62 @@ public class FileConfigurationTest extends ConfigurationImplTest {
assertEquals(123, conf.getDiskScanPeriod());
}
+ private void verifyAddresses() {
+ assertEquals(2, conf.getAddressConfigurations().size());
+
+ // Addr 1
+ CoreAddressConfiguration addressConfiguration = conf.getAddressConfigurations().get(0);
+ assertEquals("addr1", addressConfiguration.getName());
+ assertEquals(ANYCAST, addressConfiguration.getRoutingType());
+ assertEquals(2, addressConfiguration.getQueueConfigurations().size());
+
+ // Addr 1 Queue 1
+ CoreQueueConfiguration queueConfiguration = addressConfiguration.getQueueConfigurations().get(0);
+
+ assertEquals("q1", queueConfiguration.getName());
+ assertFalse(queueConfiguration.isDurable());
+ assertEquals("color='blue'", queueConfiguration.getFilterString());
+ assertEquals(addressConfiguration.getDefaultDeleteOnNoConsumers(), queueConfiguration.getDeleteOnNoConsumers());
+ assertEquals("addr1", queueConfiguration.getAddress());
+ assertEquals(addressConfiguration.getDefaultMaxConsumers(), queueConfiguration.getMaxConsumers());
+
+ // Addr 1 Queue 2
+ queueConfiguration = addressConfiguration.getQueueConfigurations().get(1);
+
+ assertEquals("q2", queueConfiguration.getName());
+ assertTrue(queueConfiguration.isDurable());
+ assertEquals("color='green'", queueConfiguration.getFilterString());
+ assertEquals(new Integer(-1), queueConfiguration.getMaxConsumers());
+ assertFalse(queueConfiguration.getDeleteOnNoConsumers());
+ assertEquals("addr1", queueConfiguration.getAddress());
+
+ // Addr 2
+ addressConfiguration = conf.getAddressConfigurations().get(1);
+ assertEquals("addr2", addressConfiguration.getName());
+ assertEquals(MULTICAST, addressConfiguration.getRoutingType());
+ assertEquals(2, addressConfiguration.getQueueConfigurations().size());
+
+ // Addr 2 Queue 1
+ queueConfiguration = addressConfiguration.getQueueConfigurations().get(0);
+
+ assertEquals("q3", queueConfiguration.getName());
+ assertTrue(queueConfiguration.isDurable());
+ assertEquals("color='red'", queueConfiguration.getFilterString());
+ assertEquals(new Integer(10), queueConfiguration.getMaxConsumers());
+ assertEquals(addressConfiguration.getDefaultDeleteOnNoConsumers(), queueConfiguration.getDeleteOnNoConsumers());
+ assertEquals("addr2", queueConfiguration.getAddress());
+
+ // Addr 2 Queue 2
+ queueConfiguration = addressConfiguration.getQueueConfigurations().get(1);
+
+ assertEquals("q4", queueConfiguration.getName());
+ assertTrue(queueConfiguration.isDurable());
+ assertNull(queueConfiguration.getFilterString());
+ assertEquals(addressConfiguration.getDefaultMaxConsumers(), queueConfiguration.getMaxConsumers());
+ assertTrue(queueConfiguration.getDeleteOnNoConsumers());
+ assertEquals("addr2", queueConfiguration.getAddress());
+ }
+
@Test
public void testSecuritySettingPlugin() throws Exception {
FileConfiguration fc = new FileConfiguration();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8c51ed73/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/resources/ConfigurationTest-full-config.xml b/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
index 8984405..3bc14bf 100644
--- a/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
+++ b/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
@@ -289,5 +289,31 @@
<factory-class>org.foo</factory-class>
</connector-service>
</connector-services>
+
+ <addresses>
+ <address name="addr1" type="anycast">
+ <queues>
+ <queue name="q1">
+ <durable>false</durable>
+ <filter string="color='blue'"/>
+ </queue>
+ <queue name="q2" max-consumers="-1" delete-on-no-consumers="false">
+ <durable>true</durable>
+ <filter string="color='green'"/>
+ </queue>
+ </queues>
+ </address>
+ <address name="addr2" type="multicast">
+ <queues>
+ <queue name="q3" max-consumers="10" >
+ <filter string="color='red'"/>
+ </queue>
+ <queue name="q4" delete-on-no-consumers="true">
+ <durable>true</durable>
+ </queue>
+ </queues>
+ </address>
+ </addresses>
+
</core>
</configuration>