You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2019/10/28 13:04:16 UTC
[activemq-artemis] branch master updated (47a5406 -> eb3c493)
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git.
from 47a5406 Merge branch 'ARTEMIS-2531'
new c0e77e9 ARTEMIS-2529 update address-settings mngmnt
new 84067d8 ARTEMIS-2504 implement retroactive addresses
new eb3c493 This closes #2850
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/activemq/artemis/logs/AuditLogger.java | 7 +
.../api/config/ActiveMQDefaultConfiguration.java | 6 +
.../api/core/management/ActiveMQServerControl.java | 58 ++-
.../api/core/management/AddressControl.java | 3 +
.../api/core/management/AddressSettingsInfo.java | 270 +++++++++++-
.../artemis/api/core/management/DivertControl.java | 3 +
.../artemis/api/core/management/QueueControl.java | 6 +
.../artemis/api/core/management/ResourceNames.java | 31 +-
.../core/management/AddressSettingsInfoTest.java | 57 ++-
.../api/core/management/ResourceNamesTest.java | 91 ++++
.../deployers/impl/FileConfigurationParser.java | 6 +
.../management/impl/ActiveMQServerControlImpl.java | 224 ++++++++--
.../core/management/impl/AddressControlImpl.java | 8 +
.../core/management/impl/DivertControlImpl.java | 15 +-
.../core/management/impl/QueueControlImpl.java | 16 +
.../core/postoffice/impl/PostOfficeImpl.java | 133 ++++++
.../core/postoffice/impl/SimpleAddressManager.java | 2 +-
.../artemis/core/server/RoutingContext.java | 2 +-
.../core/server/impl/ActiveMQServerImpl.java | 20 +-
.../artemis/core/server/impl/AddressInfo.java | 14 +-
.../artemis/core/server/impl/DivertImpl.java | 2 +-
.../artemis/core/server/impl/JournalLoader.java | 3 +-
.../core/server/impl/PostOfficeJournalLoader.java | 5 +-
.../artemis/core/server/impl/QueueImpl.java | 33 +-
.../core/server/impl/RoutingContextImpl.java | 3 +-
.../management/impl/ManagementServiceImpl.java | 2 +-
.../core/settings/impl/AddressSettings.java | 50 ++-
.../artemis/core/settings/impl/DeletionPolicy.java | 11 +
.../core/settings/impl/SlowConsumerPolicy.java | 11 +
.../resources/schema/artemis-configuration.xsd | 11 +-
.../core/config/impl/FileConfigurationTest.java | 2 +
.../impl/journal/AddressBindingEncodingTest.java | 52 +++
.../resources/ConfigurationTest-full-config.xml | 1 +
...rationTest-xinclude-config-address-settings.xml | 1 +
.../src/test/resources/artemis-configuration.xsd | 15 +
docs/user-manual/en/SUMMARY.md | 1 +
docs/user-manual/en/address-model.md | 13 +-
docs/user-manual/en/configuration-index.md | 4 +
docs/user-manual/en/retroactive-addresses.md | 88 ++++
.../RequestReplyMultiProtocolTest.java | 5 +-
.../crossprotocol/RequestReplyNonJMSTest.java | 14 +-
.../management/ActiveMQServerControlTest.java | 302 ++++++++++++-
.../ActiveMQServerControlUsingCoreTest.java | 155 ++++++-
.../integration/management/AddressControlTest.java | 13 +
.../management/AddressControlUsingCoreTest.java | 5 +
.../integration/management/DivertControlTest.java | 24 ++
.../management/DivertControlUsingCoreTest.java | 5 +
.../integration/management/QueueControlTest.java | 24 +-
.../management/QueueControlUsingCoreTest.java | 5 +
.../server/RetroactiveAddressFailoverTest.java | 119 ++++++
.../integration/server/RetroactiveAddressTest.java | 476 +++++++++++++++++++++
.../core/server/impl/fakes/FakeJournalLoader.java | 3 +-
52 files changed, 2341 insertions(+), 89 deletions(-)
create mode 100644 artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/ResourceNamesTest.java
create mode 100644 artemis-server/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/AddressBindingEncodingTest.java
create mode 100644 docs/user-manual/en/retroactive-addresses.md
create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/RetroactiveAddressFailoverTest.java
create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/RetroactiveAddressTest.java
[activemq-artemis] 01/03: ARTEMIS-2529 update address-settings
mngmnt
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit c0e77e96d1eeaf770a209353b6366c43511c1cdc
Author: Justin Bertram <jb...@apache.org>
AuthorDate: Thu Oct 24 10:53:28 2019 -0500
ARTEMIS-2529 update address-settings mngmnt
---
.../api/core/management/ActiveMQServerControl.java | 57 +++-
.../api/core/management/AddressSettingsInfo.java | 261 +++++++++++++++++-
.../core/management/AddressSettingsInfoTest.java | 55 +++-
.../management/impl/ActiveMQServerControlImpl.java | 220 ++++++++++++---
.../core/settings/impl/AddressSettings.java | 18 ++
.../artemis/core/settings/impl/DeletionPolicy.java | 11 +
.../core/settings/impl/SlowConsumerPolicy.java | 11 +
.../resources/schema/artemis-configuration.xsd | 3 +-
docs/user-manual/en/address-model.md | 3 +-
docs/user-manual/en/configuration-index.md | 3 +
.../management/ActiveMQServerControlTest.java | 296 ++++++++++++++++++++-
.../ActiveMQServerControlUsingCoreTest.java | 153 ++++++++++-
12 files changed, 1037 insertions(+), 54 deletions(-)
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java
index 4516cca..207a8cc 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java
@@ -1330,8 +1330,61 @@ public interface ActiveMQServerControl {
@Parameter(desc = "allow auto-created jms topics to be deleted automatically", name = "autoDeleteJmsTopics") boolean autoDeleteJmsTopics,
@Parameter(desc = "allow queues to be created automatically", name = "autoCreateQueues") boolean autoCreateQueues,
@Parameter(desc = "allow auto-created queues to be deleted automatically", name = "autoDeleteQueues") boolean autoDeleteQueues,
- @Parameter(desc = "allow topics to be created automatically", name = "autoCreateAddresses") boolean autoCreateAddresses,
- @Parameter(desc = "allow auto-created topics to be deleted automatically", name = "autoDeleteAddresses") boolean autoDeleteAddresses) throws Exception;
+ @Parameter(desc = "allow addresses to be created automatically", name = "autoCreateAddresses") boolean autoCreateAddresses,
+ @Parameter(desc = "allow auto-created addresses to be deleted automatically", name = "autoDeleteAddresses") boolean autoDeleteAddresses) throws Exception;
+
+ /**
+ * adds a new address setting for a specific address
+ */
+ @Operation(desc = "Add address settings for addresses matching the addressMatch", impact = MBeanOperationInfo.ACTION)
+ void addAddressSettings(@Parameter(desc = "an address match", name = "addressMatch") String addressMatch,
+ @Parameter(desc = "the dead letter address setting", name = "DLA") String DLA,
+ @Parameter(desc = "the expiry address setting", name = "expiryAddress") String expiryAddress,
+ @Parameter(desc = "the expiry delay setting", name = "expiryDelay") long expiryDelay,
+ @Parameter(desc = "are any queues created for this address a last value queue", name = "lastValueQueue") boolean lastValueQueue,
+ @Parameter(desc = "the delivery attempts", name = "deliveryAttempts") int deliveryAttempts,
+ @Parameter(desc = "the max size in bytes", name = "maxSizeBytes") long maxSizeBytes,
+ @Parameter(desc = "the page size in bytes", name = "pageSizeBytes") int pageSizeBytes,
+ @Parameter(desc = "the max number of pages in the soft memory cache", name = "pageMaxCacheSize") int pageMaxCacheSize,
+ @Parameter(desc = "the redelivery delay", name = "redeliveryDelay") long redeliveryDelay,
+ @Parameter(desc = "the redelivery delay multiplier", name = "redeliveryMultiplier") double redeliveryMultiplier,
+ @Parameter(desc = "the maximum redelivery delay", name = "maxRedeliveryDelay") long maxRedeliveryDelay,
+ @Parameter(desc = "the redistribution delay", name = "redistributionDelay") long redistributionDelay,
+ @Parameter(desc = "do we send to the DLA when there is no where to route the message", name = "sendToDLAOnNoRoute") boolean sendToDLAOnNoRoute,
+ @Parameter(desc = "the policy to use when the address is full", name = "addressFullMessagePolicy") String addressFullMessagePolicy,
+ @Parameter(desc = "when a consumer falls below this threshold in terms of messages consumed per second it will be considered 'slow'", name = "slowConsumerThreshold") long slowConsumerThreshold,
+ @Parameter(desc = "how often (in seconds) to check for slow consumers", name = "slowConsumerCheckPeriod") long slowConsumerCheckPeriod,
+ @Parameter(desc = "the policy to use when a slow consumer is detected", name = "slowConsumerPolicy") String slowConsumerPolicy,
+ @Parameter(desc = "allow jms queues to be created automatically", name = "autoCreateJmsQueues") boolean autoCreateJmsQueues,
+ @Parameter(desc = "allow auto-created jms queues to be deleted automatically", name = "autoDeleteJmsQueues") boolean autoDeleteJmsQueues,
+ @Parameter(desc = "allow jms topics to be created automatically", name = "autoCreateJmsTopics") boolean autoCreateJmsTopics,
+ @Parameter(desc = "allow auto-created jms topics to be deleted automatically", name = "autoDeleteJmsTopics") boolean autoDeleteJmsTopics,
+ @Parameter(desc = "allow queues to be created automatically", name = "autoCreateQueues") boolean autoCreateQueues,
+ @Parameter(desc = "allow auto-created queues to be deleted automatically", name = "autoDeleteQueues") boolean autoDeleteQueues,
+ @Parameter(desc = "allow addresses to be created automatically", name = "autoCreateAddresses") boolean autoCreateAddresses,
+ @Parameter(desc = "allow auto-created addresses to be deleted automatically", name = "autoDeleteAddresses") boolean autoDeleteAddresses,
+ @Parameter(desc = "how to deal with queues deleted from XML at runtime", name = "configDeleteQueues") String configDeleteQueues,
+ @Parameter(desc = "how to deal with addresses deleted from XML at runtime", name = "configDeleteAddresses") String configDeleteAddresses,
+ @Parameter(desc = "used with `BLOCK`, the max size an address can reach before messages are rejected; works in combination with `max-size-bytes` for AMQP clients only", name = "maxSizeBytesRejectThreshold") long maxSizeBytesRejectThreshold,
+ @Parameter(desc = "last-value-key value if none is set on the queue", name = "defaultLastValueKey") String defaultLastValueKey,
+ @Parameter(desc = "non-destructive value if none is set on the queue", name = "defaultNonDestructive") boolean defaultNonDestructive,
+ @Parameter(desc = "exclusive value if none is set on the queue", name = "defaultExclusiveQueue") boolean defaultExclusiveQueue,
+ @Parameter(desc = "group-rebalance value if none is set on the queue", name = "defaultGroupRebalance") boolean defaultGroupRebalance,
+ @Parameter(desc = "group-buckets value if none is set on the queue", name = "defaultGroupBuckets") int defaultGroupBuckets,
+ @Parameter(desc = "group-first-key value if none is set on the queue", name = "defaultGroupFirstKey") String defaultGroupFirstKey,
+ @Parameter(desc = "max-consumers value if none is set on the queue", name = "defaultMaxConsumers") int defaultMaxConsumers,
+ @Parameter(desc = "purge-on-no-consumers value if none is set on the queue", name = "defaultPurgeOnNoConsumers") boolean defaultPurgeOnNoConsumers,
+ @Parameter(desc = "consumers-before-dispatch value if none is set on the queue", name = "defaultConsumersBeforeDispatch") int defaultConsumersBeforeDispatch,
+ @Parameter(desc = "delay-before-dispatch value if none is set on the queue", name = "defaultDelayBeforeDispatch") long defaultDelayBeforeDispatch,
+ @Parameter(desc = "routing-type value if none is set on the queue", name = "defaultQueueRoutingType") String defaultQueueRoutingType,
+ @Parameter(desc = "routing-type value if none is set on the address", name = "defaultAddressRoutingType") String defaultAddressRoutingType,
+ @Parameter(desc = "consumer-window-size value if none is set on the queue", name = "defaultConsumerWindowSize") int defaultConsumerWindowSize,
+ @Parameter(desc = "ring-size value if none is set on the queue", name = "defaultRingSize") long defaultRingSize,
+ @Parameter(desc = "allow created queues to be deleted automatically", name = "autoDeleteCreatedQueues") boolean autoDeleteCreatedQueues,
+ @Parameter(desc = "delay for deleting auto-created queues", name = "autoDeleteQueuesDelay") long autoDeleteQueuesDelay,
+ @Parameter(desc = "the message count the queue must be at or below before it can be auto deleted", name = "autoDeleteQueuesMessageCount") long autoDeleteQueuesMessageCount,
+ @Parameter(desc = "delay for deleting auto-created addresses", name = "autoDeleteAddressesDelay") long autoDeleteAddressesDelay,
+ @Parameter(desc = "factor by which to modify the redelivery delay slightly to avoid collisions", name = "redeliveryCollisionAvoidanceFactor") double redeliveryCollisionAvoidanceFactor) throws Exception;
@Operation(desc = "Remove address settings", impact = MBeanOperationInfo.ACTION)
void removeAddressSettings(@Parameter(desc = "an address match", name = "addressMatch") String addressMatch) throws Exception;
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfo.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfo.java
index 7c2b074..53b84d5 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfo.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfo.java
@@ -63,11 +63,108 @@ public final class AddressSettingsInfo {
private final boolean autoDeleteJmsTopics;
+ private final boolean autoCreateQueues;
+
+ private final boolean autoDeleteQueues;
+
+ private final boolean autoCreateAddresses;
+
+ private final boolean autoDeleteAddresses;
+
+ private final String configDeleteQueues;
+
+ private final String configDeleteAddresses;
+
+ private final long maxSizeBytesRejectThreshold;
+
+ private final String defaultLastValueKey;
+
+ private final boolean defaultNonDestructive;
+
+ private final boolean defaultExclusiveQueue;
+
+ private final boolean defaultGroupRebalance;
+
+ private final int defaultGroupBuckets;
+
+ private final String defaultGroupFirstKey;
+
+ private final int defaultMaxConsumers;
+
+ private final boolean defaultPurgeOnNoConsumers;
+
+ private final int defaultConsumersBeforeDispatch;
+
+ private final long defaultDelayBeforeDispatch;
+
+ private final String defaultQueueRoutingType;
+
+ private final String defaultAddressRoutingType;
+
+ private final int defaultConsumerWindowSize;
+
+ private final long defaultRingSize;
+
+ private final boolean autoDeleteCreatedQueues;
+
+ private final long autoDeleteQueuesDelay;
+
+ private final long autoDeleteQueuesMessageCount;
+
+ private final long autoDeleteAddressesDelay;
+
+ private final double redeliveryCollisionAvoidanceFactor;
+
// Static --------------------------------------------------------
public static AddressSettingsInfo from(final String jsonString) {
JsonObject object = JsonUtil.readJsonObject(jsonString);
- return new AddressSettingsInfo(object.getString("addressFullMessagePolicy"), object.getJsonNumber("maxSizeBytes").longValue(), object.getInt("pageSizeBytes"), object.getInt("pageCacheMaxSize"), object.getInt("maxDeliveryAttempts"), object.getJsonNumber("redeliveryDelay").longValue(), object.getJsonNumber("redeliveryMultiplier").doubleValue(), object.getJsonNumber("maxRedeliveryDelay").longValue(), object.getString("DLA"), object.getString("expiryAddress"), object.getBoolean("lastVa [...]
+ return new AddressSettingsInfo(object.getString("addressFullMessagePolicy"),
+ object.getJsonNumber("maxSizeBytes").longValue(),
+ object.getInt("pageSizeBytes"),
+ object.getInt("pageCacheMaxSize"),
+ object.getInt("maxDeliveryAttempts"),
+ object.getJsonNumber("redeliveryDelay").longValue(),
+ object.getJsonNumber("redeliveryMultiplier").doubleValue(),
+ object.getJsonNumber("maxRedeliveryDelay").longValue(),
+ object.getString("DLA"),
+ object.getString("expiryAddress"),
+ object.getBoolean("lastValueQueue"),
+ object.getJsonNumber("redistributionDelay").longValue(),
+ object.getBoolean("sendToDLAOnNoRoute"),
+ object.getJsonNumber("slowConsumerThreshold").longValue(),
+ object.getJsonNumber("slowConsumerCheckPeriod").longValue(),
+ object.getString("slowConsumerPolicy"),
+ object.getBoolean("autoCreateJmsQueues"),
+ object.getBoolean("autoCreateJmsTopics"),
+ object.getBoolean("autoDeleteJmsQueues"),
+ object.getBoolean("autoDeleteJmsTopics"),
+ object.getBoolean("autoCreateQueues"),
+ object.getBoolean("autoDeleteQueues"),
+ object.getBoolean("autoCreateAddresses"),
+ object.getBoolean("autoDeleteAddresses"),
+ object.getString("configDeleteQueues"),
+ object.getString("configDeleteAddresses"),
+ object.getJsonNumber("maxSizeBytesRejectThreshold").longValue(),
+ object.getString("defaultLastValueKey"),
+ object.getBoolean("defaultNonDestructive"),
+ object.getBoolean("defaultExclusiveQueue"),
+ object.getBoolean("defaultGroupRebalance"),
+ object.getInt("defaultGroupBuckets"),
+ object.getString("defaultGroupFirstKey"),
+ object.getInt("defaultMaxConsumers"),
+ object.getBoolean("defaultPurgeOnNoConsumers"),
+ object.getInt("defaultConsumersBeforeDispatch"),
+ object.getJsonNumber("defaultDelayBeforeDispatch").longValue(),
+ object.getString("defaultQueueRoutingType"),
+ object.getString("defaultAddressRoutingType"),
+ object.getInt("defaultConsumerWindowSize"),
+ object.getJsonNumber("defaultRingSize").longValue(),
+ object.getBoolean("autoDeleteCreatedQueues"),
+ object.getJsonNumber("autoDeleteQueuesDelay").longValue(),
+ object.getJsonNumber("autoDeleteQueuesMessageCount").longValue(),
+ object.getJsonNumber("autoDeleteAddressesDelay").longValue(),
+ object.getJsonNumber("redeliveryCollisionAvoidanceFactor").doubleValue());
}
// Constructors --------------------------------------------------
@@ -91,7 +188,33 @@ public final class AddressSettingsInfo {
boolean autoCreateJmsQueues,
boolean autoCreateJmsTopics,
boolean autoDeleteJmsQueues,
- boolean autoDeleteJmsTopics) {
+ boolean autoDeleteJmsTopics,
+ boolean autoCreateQueues,
+ boolean autoDeleteQueues,
+ boolean autoCreateAddresses,
+ boolean autoDeleteAddresses,
+ String configDeleteQueues,
+ String configDeleteAddresses,
+ long maxSizeBytesRejectThreshold,
+ String defaultLastValueKey,
+ boolean defaultNonDestructive,
+ boolean defaultExclusiveQueue,
+ boolean defaultGroupRebalance,
+ int defaultGroupBuckets,
+ String defaultGroupFirstKey,
+ int defaultMaxConsumers,
+ boolean defaultPurgeOnNoConsumers,
+ int defaultConsumersBeforeDispatch,
+ long defaultDelayBeforeDispatch,
+ String defaultQueueRoutingType,
+ String defaultAddressRoutingType,
+ int defaultConsumerWindowSize,
+ long defaultRingSize,
+ boolean autoDeleteCreatedQueues,
+ long autoDeleteQueuesDelay,
+ long autoDeleteQueuesMessageCount,
+ long autoDeleteAddressesDelay,
+ double redeliveryCollisionAvoidanceFactor) {
this.addressFullMessagePolicy = addressFullMessagePolicy;
this.maxSizeBytes = maxSizeBytes;
this.pageSizeBytes = pageSizeBytes;
@@ -112,6 +235,32 @@ public final class AddressSettingsInfo {
this.autoDeleteJmsQueues = autoDeleteJmsQueues;
this.autoCreateJmsTopics = autoCreateJmsTopics;
this.autoDeleteJmsTopics = autoDeleteJmsTopics;
+ this.autoCreateQueues = autoCreateQueues;
+ this.autoDeleteQueues = autoDeleteQueues;
+ this.autoCreateAddresses = autoCreateAddresses;
+ this.autoDeleteAddresses = autoDeleteAddresses;
+ this.configDeleteQueues = configDeleteQueues;
+ this.configDeleteAddresses = configDeleteAddresses;
+ this.maxSizeBytesRejectThreshold = maxSizeBytesRejectThreshold;
+ this.defaultLastValueKey = defaultLastValueKey;
+ this.defaultNonDestructive = defaultNonDestructive;
+ this.defaultExclusiveQueue = defaultExclusiveQueue;
+ this.defaultGroupRebalance = defaultGroupRebalance;
+ this.defaultGroupBuckets = defaultGroupBuckets;
+ this.defaultGroupFirstKey = defaultGroupFirstKey;
+ this.defaultMaxConsumers = defaultMaxConsumers;
+ this.defaultPurgeOnNoConsumers = defaultPurgeOnNoConsumers;
+ this.defaultConsumersBeforeDispatch = defaultConsumersBeforeDispatch;
+ this.defaultDelayBeforeDispatch = defaultDelayBeforeDispatch;
+ this.defaultQueueRoutingType = defaultQueueRoutingType;
+ this.defaultAddressRoutingType = defaultAddressRoutingType;
+ this.defaultConsumerWindowSize = defaultConsumerWindowSize;
+ this.defaultRingSize = defaultRingSize;
+ this.autoDeleteCreatedQueues = autoDeleteCreatedQueues;
+ this.autoDeleteQueuesDelay = autoDeleteQueuesDelay;
+ this.autoDeleteQueuesMessageCount = autoDeleteQueuesMessageCount;
+ this.autoDeleteAddressesDelay = autoDeleteAddressesDelay;
+ this.redeliveryCollisionAvoidanceFactor = redeliveryCollisionAvoidanceFactor;
}
// Public --------------------------------------------------------
@@ -184,20 +333,128 @@ public final class AddressSettingsInfo {
return slowConsumerPolicy;
}
+ @Deprecated
public boolean isAutoCreateJmsQueues() {
return autoCreateJmsQueues;
}
+ @Deprecated
public boolean isAutoDeleteJmsQueues() {
return autoDeleteJmsQueues;
}
+ @Deprecated
public boolean isAutoCreateJmsTopics() {
return autoCreateJmsTopics;
}
+ @Deprecated
public boolean isAutoDeleteJmsTopics() {
return autoDeleteJmsTopics;
}
+
+ public boolean isAutoCreateQueues() {
+ return autoCreateQueues;
+ }
+
+ public boolean isAutoDeleteQueues() {
+ return autoDeleteQueues;
+ }
+
+ public boolean isAutoCreateAddresses() {
+ return autoCreateAddresses;
+ }
+
+ public boolean isAutoDeleteAddresses() {
+ return autoDeleteAddresses;
+ }
+
+ public String getConfigDeleteQueues() {
+ return configDeleteQueues;
+ }
+
+ public String getConfigDeleteAddresses() {
+ return configDeleteAddresses;
+ }
+
+ public long getMaxSizeBytesRejectThreshold() {
+ return maxSizeBytesRejectThreshold;
+ }
+
+ public String getDefaultLastValueKey() {
+ return defaultLastValueKey;
+ }
+
+ public boolean isDefaultNonDestructive() {
+ return defaultNonDestructive;
+ }
+
+ public boolean isDefaultExclusiveQueue() {
+ return defaultExclusiveQueue;
+ }
+
+ public boolean isDefaultGroupRebalance() {
+ return defaultGroupRebalance;
+ }
+
+ public int getDefaultGroupBuckets() {
+ return defaultGroupBuckets;
+ }
+
+ public String getDefaultGroupFirstKey() {
+ return defaultGroupFirstKey;
+ }
+
+ public int getDefaultMaxConsumers() {
+ return defaultMaxConsumers;
+ }
+
+ public boolean isDefaultPurgeOnNoConsumers() {
+ return defaultPurgeOnNoConsumers;
+ }
+
+ public int getDefaultConsumersBeforeDispatch() {
+ return defaultConsumersBeforeDispatch;
+ }
+
+ public long getDefaultDelayBeforeDispatch() {
+ return defaultDelayBeforeDispatch;
+ }
+
+ public String getDefaultQueueRoutingType() {
+ return defaultQueueRoutingType;
+ }
+
+ public String getDefaultAddressRoutingType() {
+ return defaultAddressRoutingType;
+ }
+
+ public int getDefaultConsumerWindowSize() {
+ return defaultConsumerWindowSize;
+ }
+
+ public long getDefaultRingSize() {
+ return defaultRingSize;
+ }
+
+ public boolean isAutoDeleteCreatedQueues() {
+ return autoDeleteCreatedQueues;
+ }
+
+ public long getAutoDeleteQueuesDelay() {
+ return autoDeleteQueuesDelay;
+ }
+
+ public long getAutoDeleteQueuesMessageCount() {
+ return autoDeleteQueuesMessageCount;
+ }
+
+ public long getAutoDeleteAddressesDelay() {
+ return autoDeleteAddressesDelay;
+ }
+
+ public double getRedeliveryCollisionAvoidanceFactor() {
+ return redeliveryCollisionAvoidanceFactor;
+ }
}
diff --git a/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfoTest.java b/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfoTest.java
index 392231b..65f16fa 100644
--- a/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfoTest.java
+++ b/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfoTest.java
@@ -19,6 +19,7 @@
package org.apache.activemq.artemis.api.core.management;
+import org.apache.activemq.artemis.api.core.RoutingType;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -49,7 +50,33 @@ public class AddressSettingsInfoTest {
"\"autoCreateJmsQueues\":true,\n" +
"\"autoDeleteJmsQueues\":false,\n" +
"\"autoCreateJmsTopics\":true,\n" +
- "\"autoDeleteJmsTopics\":false\n" +
+ "\"autoDeleteJmsTopics\":false,\n" +
+ "\"autoCreateQueues\":false,\n" +
+ "\"autoDeleteQueues\":false,\n" +
+ "\"autoCreateAddresses\":false,\n" +
+ "\"autoDeleteAddresses\":false,\n" +
+ "\"configDeleteQueues\":\"OFF\",\n" +
+ "\"configDeleteAddresses\":\"FORCE\",\n" +
+ "\"maxSizeBytesRejectThreshold\":1023,\n" +
+ "\"defaultLastValueKey\":\"yyy\",\n" +
+ "\"defaultNonDestructive\":false,\n" +
+ "\"defaultExclusiveQueue\":false,\n" +
+ "\"defaultGroupRebalance\":false,\n" +
+ "\"defaultGroupBuckets\":1026,\n" +
+ "\"defaultGroupFirstKey\":\"xxx\",\n" +
+ "\"defaultMaxConsumers\":1001,\n" +
+ "\"defaultPurgeOnNoConsumers\":false,\n" +
+ "\"defaultConsumersBeforeDispatch\":1005,\n" +
+ "\"defaultDelayBeforeDispatch\":1003,\n" +
+ "\"defaultQueueRoutingType\":\"MULTICAST\",\n" +
+ "\"defaultAddressRoutingType\":\"ANYCAST\",\n" +
+ "\"defaultConsumerWindowSize\":2001,\n" +
+ "\"defaultRingSize\":999,\n" +
+ "\"autoDeleteCreatedQueues\":false,\n" +
+ "\"autoDeleteQueuesDelay\":4,\n" +
+ "\"autoDeleteQueuesMessageCount\":8,\n" +
+ "\"autoDeleteAddressesDelay\":3003,\n" +
+ "\"redeliveryCollisionAvoidanceFactor\":1.1\n" +
"}";
AddressSettingsInfo addressSettingsInfo = AddressSettingsInfo.from(json);
assertEquals("fullPolicy", addressSettingsInfo.getAddressFullMessagePolicy());
@@ -72,6 +99,32 @@ public class AddressSettingsInfoTest {
assertTrue(addressSettingsInfo.isAutoCreateJmsTopics());
assertFalse(addressSettingsInfo.isAutoDeleteJmsQueues());
assertFalse(addressSettingsInfo.isAutoDeleteJmsTopics());
+ assertFalse(addressSettingsInfo.isAutoCreateQueues());
+ assertFalse(addressSettingsInfo.isAutoDeleteQueues());
+ assertFalse(addressSettingsInfo.isAutoCreateAddresses());
+ assertFalse(addressSettingsInfo.isAutoDeleteAddresses());
+ assertEquals("OFF", addressSettingsInfo.getConfigDeleteQueues());
+ assertEquals("FORCE", addressSettingsInfo.getConfigDeleteAddresses());
+ assertEquals(1023, addressSettingsInfo.getMaxSizeBytesRejectThreshold());
+ assertEquals("yyy", addressSettingsInfo.getDefaultLastValueKey());
+ assertFalse(addressSettingsInfo.isDefaultNonDestructive());
+ assertFalse(addressSettingsInfo.isDefaultExclusiveQueue());
+ assertFalse(addressSettingsInfo.isDefaultGroupRebalance());
+ assertEquals(1026, addressSettingsInfo.getDefaultGroupBuckets());
+ assertEquals("xxx", addressSettingsInfo.getDefaultGroupFirstKey());
+ assertEquals(1001, addressSettingsInfo.getDefaultMaxConsumers());
+ assertFalse(addressSettingsInfo.isDefaultPurgeOnNoConsumers());
+ assertEquals(1005, addressSettingsInfo.getDefaultConsumersBeforeDispatch());
+ assertEquals(1003, addressSettingsInfo.getDefaultDelayBeforeDispatch());
+ assertEquals(RoutingType.MULTICAST.toString(), addressSettingsInfo.getDefaultQueueRoutingType());
+ assertEquals(RoutingType.ANYCAST.toString(), addressSettingsInfo.getDefaultAddressRoutingType());
+ assertEquals(2001, addressSettingsInfo.getDefaultConsumerWindowSize());
+ assertEquals(999, addressSettingsInfo.getDefaultRingSize());
+ assertFalse(addressSettingsInfo.isAutoDeleteCreatedQueues());
+ assertEquals(4, addressSettingsInfo.getAutoDeleteQueuesDelay());
+ assertEquals(8, addressSettingsInfo.getAutoDeleteQueuesMessageCount());
+ assertEquals(3003, addressSettingsInfo.getAutoDeleteAddressesDelay());
+ assertEquals(1.1, addressSettingsInfo.getRedeliveryCollisionAvoidanceFactor(), 0);
}
}
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 4c8c78a..ae94afc 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
@@ -54,6 +54,7 @@ import org.apache.activemq.artemis.api.core.JsonUtil;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
+import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
import org.apache.activemq.artemis.api.core.management.ActiveMQServerControl;
import org.apache.activemq.artemis.api.core.management.AddressControl;
import org.apache.activemq.artemis.api.core.management.BridgeControl;
@@ -112,6 +113,7 @@ import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.server.impl.SharedNothingLiveActivation;
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.DeletionPolicy;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.apache.activemq.artemis.core.transaction.ResourceManager;
import org.apache.activemq.artemis.core.transaction.Transaction;
@@ -2721,13 +2723,35 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
.add("slowConsumerCheckPeriod", addressSettings.getSlowConsumerCheckPeriod())
.add("slowConsumerPolicy", consumerPolicy)
.add("autoCreateJmsQueues", addressSettings.isAutoCreateJmsQueues())
- .add("autoCreateJmsTopics", addressSettings.isAutoCreateJmsTopics())
.add("autoDeleteJmsQueues", addressSettings.isAutoDeleteJmsQueues())
- .add("autoDeleteJmsTopics", addressSettings.isAutoDeleteJmsQueues())
+ .add("autoCreateJmsTopics", addressSettings.isAutoCreateJmsTopics())
+ .add("autoDeleteJmsTopics", addressSettings.isAutoDeleteJmsTopics())
.add("autoCreateQueues", addressSettings.isAutoCreateQueues())
.add("autoDeleteQueues", addressSettings.isAutoDeleteQueues())
- .add("autoCreateAddress", addressSettings.isAutoCreateAddresses())
- .add("autoDeleteAddress", addressSettings.isAutoDeleteAddresses())
+ .add("autoCreateAddresses", addressSettings.isAutoCreateAddresses())
+ .add("autoDeleteAddresses", addressSettings.isAutoDeleteAddresses())
+ .add("configDeleteQueues", addressSettings.getConfigDeleteQueues().toString())
+ .add("configDeleteAddresses", addressSettings.getConfigDeleteAddresses().toString())
+ .add("maxSizeBytesRejectThreshold", addressSettings.getMaxSizeBytesRejectThreshold())
+ .add("defaultLastValueKey", addressSettings.getDefaultLastValueKey() == null ? "" : addressSettings.getDefaultLastValueKey().toString())
+ .add("defaultNonDestructive", addressSettings.isDefaultNonDestructive())
+ .add("defaultExclusiveQueue", addressSettings.isDefaultExclusiveQueue())
+ .add("defaultGroupRebalance", addressSettings.isDefaultGroupRebalance())
+ .add("defaultGroupBuckets", addressSettings.getDefaultGroupBuckets())
+ .add("defaultGroupFirstKey", addressSettings.getDefaultGroupFirstKey() == null ? "" : addressSettings.getDefaultGroupFirstKey().toString())
+ .add("defaultMaxConsumers", addressSettings.getDefaultMaxConsumers())
+ .add("defaultPurgeOnNoConsumers", addressSettings.isDefaultPurgeOnNoConsumers())
+ .add("defaultConsumersBeforeDispatch", addressSettings.getDefaultConsumersBeforeDispatch())
+ .add("defaultDelayBeforeDispatch", addressSettings.getDefaultDelayBeforeDispatch())
+ .add("defaultQueueRoutingType", addressSettings.getDefaultQueueRoutingType().toString())
+ .add("defaultAddressRoutingType", addressSettings.getDefaultAddressRoutingType().toString())
+ .add("defaultConsumerWindowSize", addressSettings.getDefaultConsumerWindowSize())
+ .add("defaultRingSize", addressSettings.getDefaultRingSize())
+ .add("autoDeleteCreatedQueues", addressSettings.isAutoDeleteCreatedQueues())
+ .add("autoDeleteQueuesDelay", addressSettings.getAutoDeleteQueuesDelay())
+ .add("autoDeleteQueuesMessageCount", addressSettings.getAutoDeleteQueuesMessageCount())
+ .add("autoDeleteAddressesDelay", addressSettings.getAutoDeleteAddressesDelay())
+ .add("redeliveryCollisionAvoidanceFactor", addressSettings.getRedeliveryCollisionAvoidanceFactor())
.build()
.toString();
}
@@ -2755,7 +2779,32 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
final boolean autoDeleteJmsQueues,
final boolean autoCreateJmsTopics,
final boolean autoDeleteJmsTopics) throws Exception {
- addAddressSettings(address, DLA, expiryAddress, expiryDelay, lastValueQueue, deliveryAttempts, maxSizeBytes, pageSizeBytes, pageMaxCacheSize, redeliveryDelay, redeliveryMultiplier, maxRedeliveryDelay, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy, slowConsumerThreshold, slowConsumerCheckPeriod, slowConsumerPolicy, autoCreateJmsQueues, autoDeleteJmsQueues, autoCreateJmsTopics, autoDeleteJmsTopics, AddressSettings.DEFAULT_AUTO_CREATE_QUEUES, AddressSettings.DEFAUL [...]
+ addAddressSettings(address,
+ DLA,
+ expiryAddress,
+ expiryDelay,
+ lastValueQueue,
+ deliveryAttempts,
+ maxSizeBytes,
+ pageSizeBytes,
+ pageMaxCacheSize,
+ redeliveryDelay,
+ redeliveryMultiplier,
+ maxRedeliveryDelay,
+ redistributionDelay,
+ sendToDLAOnNoRoute,
+ addressFullMessagePolicy,
+ slowConsumerThreshold,
+ slowConsumerCheckPeriod,
+ slowConsumerPolicy,
+ autoCreateJmsQueues,
+ autoDeleteJmsQueues,
+ autoCreateJmsTopics,
+ autoDeleteJmsTopics,
+ AddressSettings.DEFAULT_AUTO_CREATE_QUEUES,
+ AddressSettings.DEFAULT_AUTO_DELETE_QUEUES,
+ AddressSettings.DEFAULT_AUTO_CREATE_ADDRESSES,
+ AddressSettings.DEFAULT_AUTO_DELETE_ADDRESSES);
}
@Override
@@ -2763,8 +2812,8 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
final String DLA,
final String expiryAddress,
final long expiryDelay,
- final boolean lastValueQueue,
- final int deliveryAttempts,
+ final boolean defaultLastValueQueue,
+ final int maxDeliveryAttempts,
final long maxSizeBytes,
final int pageSizeBytes,
final int pageMaxCacheSize,
@@ -2785,13 +2834,117 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
final boolean autoDeleteQueues,
final boolean autoCreateAddresses,
final boolean autoDeleteAddresses) throws Exception {
- if (AuditLogger.isEnabled()) {
- AuditLogger.addAddressSettings(this.server, address, DLA, expiryAddress, expiryDelay, lastValueQueue, deliveryAttempts,
+ addAddressSettings(address,
+ DLA,
+ expiryAddress,
+ expiryDelay,
+ defaultLastValueQueue,
+ maxDeliveryAttempts,
+ maxSizeBytes,
+ pageSizeBytes,
+ pageMaxCacheSize,
+ redeliveryDelay,
+ redeliveryMultiplier,
+ maxRedeliveryDelay,
+ redistributionDelay,
+ sendToDLAOnNoRoute,
+ addressFullMessagePolicy,
+ slowConsumerThreshold,
+ slowConsumerCheckPeriod,
+ slowConsumerPolicy,
+ autoCreateJmsQueues,
+ autoDeleteJmsQueues,
+ autoCreateJmsTopics,
+ autoDeleteJmsTopics,
+ autoCreateQueues,
+ autoDeleteQueues,
+ autoCreateAddresses,
+ autoDeleteAddresses,
+ AddressSettings.DEFAULT_CONFIG_DELETE_QUEUES.toString(),
+ AddressSettings.DEFAULT_CONFIG_DELETE_ADDRESSES.toString(),
+ AddressSettings.DEFAULT_ADDRESS_REJECT_THRESHOLD,
+ ActiveMQDefaultConfiguration.getDefaultLastValueKey().toString(),
+ ActiveMQDefaultConfiguration.getDefaultNonDestructive(),
+ ActiveMQDefaultConfiguration.getDefaultExclusive(),
+ ActiveMQDefaultConfiguration.getDefaultGroupRebalance(),
+ ActiveMQDefaultConfiguration.getDefaultGroupBuckets(),
+ ActiveMQDefaultConfiguration.getDefaultGroupFirstKey().toString(),
+ ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(),
+ ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(),
+ ActiveMQDefaultConfiguration.getDefaultConsumersBeforeDispatch(),
+ ActiveMQDefaultConfiguration.getDefaultDelayBeforeDispatch(),
+ ActiveMQDefaultConfiguration.getDefaultRoutingType().toString(),
+ ActiveMQDefaultConfiguration.getDefaultRoutingType().toString(),
+ ActiveMQClient.DEFAULT_CONSUMER_WINDOW_SIZE,
+ ActiveMQDefaultConfiguration.getDefaultRingSize(),
+ AddressSettings.DEFAULT_AUTO_DELETE_CREATED_QUEUES,
+ AddressSettings.DEFAULT_AUTO_DELETE_QUEUES_DELAY,
+ AddressSettings.DEFAULT_AUTO_DELETE_QUEUES_MESSAGE_COUNT,
+ AddressSettings.DEFAULT_AUTO_DELETE_ADDRESSES_DELAY,
+ AddressSettings.DEFAULT_REDELIVER_COLLISION_AVOIDANCE_FACTOR);
+ }
+
+ @Override
+ public void addAddressSettings(final String address,
+ final String DLA,
+ final String expiryAddress,
+ final long expiryDelay,
+ final boolean defaultLastValueQueue,
+ final int maxDeliveryAttempts,
+ final long maxSizeBytes,
+ final int pageSizeBytes,
+ final int pageMaxCacheSize,
+ final long redeliveryDelay,
+ final double redeliveryMultiplier,
+ final long maxRedeliveryDelay,
+ final long redistributionDelay,
+ final boolean sendToDLAOnNoRoute,
+ final String addressFullMessagePolicy,
+ final long slowConsumerThreshold,
+ final long slowConsumerCheckPeriod,
+ final String slowConsumerPolicy,
+ final boolean autoCreateJmsQueues,
+ final boolean autoDeleteJmsQueues,
+ final boolean autoCreateJmsTopics,
+ final boolean autoDeleteJmsTopics,
+ final boolean autoCreateQueues,
+ final boolean autoDeleteQueues,
+ final boolean autoCreateAddresses,
+ final boolean autoDeleteAddresses,
+ final String configDeleteQueues,
+ final String configDeleteAddresses,
+ final long maxSizeBytesRejectThreshold,
+ final String defaultLastValueKey,
+ final boolean defaultNonDestructive,
+ final boolean defaultExclusiveQueue,
+ final boolean defaultGroupRebalance,
+ final int defaultGroupBuckets,
+ final String defaultGroupFirstKey,
+ final int defaultMaxConsumers,
+ final boolean defaultPurgeOnNoConsumers,
+ final int defaultConsumersBeforeDispatch,
+ final long defaultDelayBeforeDispatch,
+ final String defaultQueueRoutingType,
+ final String defaultAddressRoutingType,
+ final int defaultConsumerWindowSize,
+ final long defaultRingSize,
+ final boolean autoDeleteCreatedQueues,
+ final long autoDeleteQueuesDelay,
+ final long autoDeleteQueuesMessageCount,
+ final long autoDeleteAddressesDelay,
+ final double redeliveryCollisionAvoidanceFactor) throws Exception {
+ if (AuditLogger.isEnabled()) {
+ AuditLogger.addAddressSettings(this.server, address, DLA, expiryAddress, expiryDelay, defaultLastValueQueue, maxDeliveryAttempts,
maxSizeBytes, pageSizeBytes, pageMaxCacheSize, redeliveryDelay, redeliveryMultiplier,
maxRedeliveryDelay, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy,
slowConsumerThreshold, slowConsumerCheckPeriod, slowConsumerPolicy, autoCreateJmsQueues,
autoDeleteJmsQueues, autoCreateJmsTopics, autoDeleteJmsTopics, autoCreateQueues, autoDeleteQueues,
- autoCreateAddresses, autoDeleteAddresses);
+ autoCreateAddresses, autoDeleteAddresses, configDeleteQueues, configDeleteAddresses, maxSizeBytesRejectThreshold,
+ defaultLastValueKey, defaultNonDestructive, defaultExclusiveQueue, defaultGroupRebalance, defaultGroupBuckets,
+ defaultGroupFirstKey, defaultMaxConsumers, defaultPurgeOnNoConsumers, defaultConsumersBeforeDispatch,
+ defaultDelayBeforeDispatch, defaultQueueRoutingType, defaultAddressRoutingType, defaultConsumerWindowSize,
+ defaultRingSize, autoDeleteCreatedQueues, autoDeleteQueuesDelay, autoDeleteQueuesMessageCount,
+ autoDeleteAddressesDelay, redeliveryCollisionAvoidanceFactor);
}
checkStarted();
@@ -2808,8 +2961,8 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
addressSettings.setDeadLetterAddress(DLA == null ? null : new SimpleString(DLA));
addressSettings.setExpiryAddress(expiryAddress == null ? null : new SimpleString(expiryAddress));
addressSettings.setExpiryDelay(expiryDelay);
- addressSettings.setDefaultLastValueQueue(lastValueQueue);
- addressSettings.setMaxDeliveryAttempts(deliveryAttempts);
+ addressSettings.setDefaultLastValueQueue(defaultLastValueQueue);
+ addressSettings.setMaxDeliveryAttempts(maxDeliveryAttempts);
addressSettings.setPageCacheMaxSize(pageMaxCacheSize);
addressSettings.setMaxSizeBytes(maxSizeBytes);
addressSettings.setPageSizeBytes(pageSizeBytes);
@@ -2818,26 +2971,10 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
addressSettings.setMaxRedeliveryDelay(maxRedeliveryDelay);
addressSettings.setRedistributionDelay(redistributionDelay);
addressSettings.setSendToDLAOnNoRoute(sendToDLAOnNoRoute);
- if (addressFullMessagePolicy == null) {
- addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
- } else if (addressFullMessagePolicy.equalsIgnoreCase("PAGE")) {
- addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
- } else if (addressFullMessagePolicy.equalsIgnoreCase("DROP")) {
- addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.DROP);
- } else if (addressFullMessagePolicy.equalsIgnoreCase("BLOCK")) {
- addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK);
- } else if (addressFullMessagePolicy.equalsIgnoreCase("FAIL")) {
- addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL);
- }
+ addressSettings.setAddressFullMessagePolicy(addressFullMessagePolicy == null ? AddressSettings.DEFAULT_ADDRESS_FULL_MESSAGE_POLICY : AddressFullMessagePolicy.valueOf(addressFullMessagePolicy.toUpperCase()));
addressSettings.setSlowConsumerThreshold(slowConsumerThreshold);
addressSettings.setSlowConsumerCheckPeriod(slowConsumerCheckPeriod);
- if (slowConsumerPolicy == null) {
- addressSettings.setSlowConsumerPolicy(SlowConsumerPolicy.NOTIFY);
- } else if (slowConsumerPolicy.equalsIgnoreCase("NOTIFY")) {
- addressSettings.setSlowConsumerPolicy(SlowConsumerPolicy.NOTIFY);
- } else if (slowConsumerPolicy.equalsIgnoreCase("KILL")) {
- addressSettings.setSlowConsumerPolicy(SlowConsumerPolicy.KILL);
- }
+ addressSettings.setSlowConsumerPolicy(slowConsumerPolicy == null ? AddressSettings.DEFAULT_SLOW_CONSUMER_POLICY : SlowConsumerPolicy.valueOf(slowConsumerPolicy.toUpperCase()));
addressSettings.setAutoCreateJmsQueues(autoCreateJmsQueues);
addressSettings.setAutoDeleteJmsQueues(autoDeleteJmsQueues);
addressSettings.setAutoCreateJmsTopics(autoCreateJmsTopics);
@@ -2846,6 +2983,29 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
addressSettings.setAutoDeleteQueues(autoDeleteQueues);
addressSettings.setAutoCreateAddresses(autoCreateAddresses);
addressSettings.setAutoDeleteAddresses(autoDeleteAddresses);
+ addressSettings.setConfigDeleteQueues(configDeleteQueues == null ? AddressSettings.DEFAULT_CONFIG_DELETE_QUEUES : DeletionPolicy.valueOf(configDeleteQueues.toUpperCase()));
+ addressSettings.setConfigDeleteAddresses(configDeleteAddresses == null ? AddressSettings.DEFAULT_CONFIG_DELETE_ADDRESSES : DeletionPolicy.valueOf(configDeleteAddresses.toUpperCase()));
+ addressSettings.setMaxSizeBytesRejectThreshold(maxSizeBytesRejectThreshold);
+ addressSettings.setDefaultLastValueKey(defaultLastValueKey == null ? ActiveMQDefaultConfiguration.getDefaultLastValueKey() : new SimpleString(defaultLastValueKey));
+ addressSettings.setDefaultNonDestructive(defaultNonDestructive);
+ addressSettings.setDefaultExclusiveQueue(defaultExclusiveQueue);
+ addressSettings.setDefaultGroupRebalance(defaultGroupRebalance);
+ addressSettings.setDefaultGroupBuckets(defaultGroupBuckets);
+ addressSettings.setDefaultGroupFirstKey(defaultGroupFirstKey == null ? ActiveMQDefaultConfiguration.getDefaultGroupFirstKey() : new SimpleString(defaultGroupFirstKey));
+ addressSettings.setDefaultMaxConsumers(defaultMaxConsumers);
+ addressSettings.setDefaultPurgeOnNoConsumers(defaultPurgeOnNoConsumers);
+ addressSettings.setDefaultConsumersBeforeDispatch(defaultConsumersBeforeDispatch);
+ addressSettings.setDefaultDelayBeforeDispatch(defaultDelayBeforeDispatch);
+ addressSettings.setDefaultQueueRoutingType(defaultQueueRoutingType == null ? ActiveMQDefaultConfiguration.getDefaultRoutingType() : RoutingType.valueOf(defaultQueueRoutingType.toUpperCase()));
+ addressSettings.setDefaultAddressRoutingType(defaultAddressRoutingType == null ? ActiveMQDefaultConfiguration.getDefaultRoutingType() : RoutingType.valueOf(defaultAddressRoutingType.toUpperCase()));
+ addressSettings.setDefaultConsumerWindowSize(defaultConsumerWindowSize);
+ addressSettings.setDefaultRingSize(defaultRingSize);
+ addressSettings.setAutoDeleteCreatedQueues(autoDeleteCreatedQueues);
+ addressSettings.setAutoDeleteQueuesDelay(autoDeleteQueuesDelay);
+ addressSettings.setAutoDeleteQueuesMessageCount(autoDeleteQueuesMessageCount);
+ addressSettings.setAutoDeleteAddressesDelay(autoDeleteAddressesDelay);
+ addressSettings.setRedeliveryCollisionAvoidanceFactor(redeliveryCollisionAvoidanceFactor);
+
server.getAddressSettingsRepository().addMatch(address, addressSettings);
storageManager.storeAddressSetting(new PersistedAddressSetting(new SimpleString(address), addressSettings));
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java
index 0129ec7..6e61683 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java
@@ -913,6 +913,9 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
if (defaultGroupBuckets == null) {
defaultGroupBuckets = merged.defaultGroupBuckets;
}
+ if (defaultGroupFirstKey == null) {
+ defaultGroupFirstKey = merged.defaultGroupFirstKey;
+ }
if (defaultRingSize == null) {
defaultRingSize = merged.defaultRingSize;
}
@@ -1080,6 +1083,10 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
if (buffer.readableBytes() > 0) {
redeliveryCollisionAvoidanceFactor = BufferHelper.readNullableDouble(buffer);
}
+
+ if (buffer.readableBytes() > 0) {
+ defaultGroupFirstKey = buffer.readNullableSimpleString();
+ }
}
@Override
@@ -1129,6 +1136,7 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
BufferHelper.sizeOfNullableLong(autoDeleteAddressesDelay) +
BufferHelper.sizeOfNullableBoolean(defaultGroupRebalance) +
BufferHelper.sizeOfNullableInteger(defaultGroupBuckets) +
+ SimpleString.sizeofNullableString(defaultGroupFirstKey) +
BufferHelper.sizeOfNullableLong(autoDeleteQueuesMessageCount) +
BufferHelper.sizeOfNullableBoolean(autoDeleteCreatedQueues) +
BufferHelper.sizeOfNullableLong(defaultRingSize);
@@ -1234,6 +1242,7 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
BufferHelper.writeNullableDouble(buffer, redeliveryCollisionAvoidanceFactor);
+ buffer.writeNullableSimpleString(defaultGroupFirstKey);
}
/* (non-Javadoc)
@@ -1292,6 +1301,7 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
result = prime * result + ((defaultConsumerWindowSize == null) ? 0 : defaultConsumerWindowSize.hashCode());
result = prime * result + ((defaultGroupRebalance == null) ? 0 : defaultGroupRebalance.hashCode());
result = prime * result + ((defaultGroupBuckets == null) ? 0 : defaultGroupBuckets.hashCode());
+ result = prime * result + ((defaultGroupFirstKey == null) ? 0 : defaultGroupFirstKey.hashCode());
result = prime * result + ((defaultRingSize == null) ? 0 : defaultRingSize.hashCode());
return result;
}
@@ -1564,6 +1574,12 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
} else if (!defaultGroupBuckets.equals(other.defaultGroupBuckets))
return false;
+ if (defaultGroupFirstKey == null) {
+ if (other.defaultGroupFirstKey != null)
+ return false;
+ } else if (!defaultGroupFirstKey.equals(other.defaultGroupFirstKey))
+ return false;
+
if (defaultRingSize == null) {
if (other.defaultRingSize != null)
return false;
@@ -1672,6 +1688,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
defaultGroupRebalance +
", defaultGroupBuckets=" +
defaultGroupBuckets +
+ ", defaultGroupFirstKey=" +
+ defaultGroupFirstKey +
", defaultRingSize=" +
defaultRingSize +
"]";
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/DeletionPolicy.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/DeletionPolicy.java
index 3ff1041..728b80f 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/DeletionPolicy.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/DeletionPolicy.java
@@ -18,4 +18,15 @@ package org.apache.activemq.artemis.core.settings.impl;
public enum DeletionPolicy {
OFF, FORCE;
+
+ public static DeletionPolicy getType(int type) {
+ switch (type) {
+ case 0:
+ return OFF;
+ case 1:
+ return FORCE;
+ default:
+ return null;
+ }
+ }
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/SlowConsumerPolicy.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/SlowConsumerPolicy.java
index 3f009d6..2e03482 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/SlowConsumerPolicy.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/SlowConsumerPolicy.java
@@ -18,4 +18,15 @@ package org.apache.activemq.artemis.core.settings.impl;
public enum SlowConsumerPolicy {
KILL, NOTIFY;
+
+ public static SlowConsumerPolicy getType(int type) {
+ switch (type) {
+ case 0:
+ return KILL;
+ case 1:
+ return NOTIFY;
+ default:
+ return null;
+ }
+ }
}
diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
index fea5cc6..529c39c 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -3278,7 +3278,8 @@
<xsd:element name="auto-delete-queues-message-count" type="xsd:long" default="0" maxOccurs="1" minOccurs="0">
<xsd:annotation>
<xsd:documentation>
- the message count the queue must be at or below before it can be evaluated to be auto deleted, 0 waits until empty queue (default) and -1 disables this check. </xsd:documentation>
+ the message count the queue must be at or below before it can be evaluated to be auto deleted, 0 waits until empty queue (default) and -1 disables this check.
+ </xsd:documentation>
</xsd:annotation>
</xsd:element>
diff --git a/docs/user-manual/en/address-model.md b/docs/user-manual/en/address-model.md
index 6b2b27e..c19da8a 100644
--- a/docs/user-manual/en/address-model.md
+++ b/docs/user-manual/en/address-model.md
@@ -583,7 +583,8 @@ that would be found in the `broker.xml` file.
<address-full-policy>PAGE</address-full-policy>
<message-counter-history-day-limit></message-counter-history-day-limit>
<last-value-queue>true</last-value-queue> <!-- deprecated! see default-last-value-queue -->
- <default-last-value-queue>true</default-last-value-queue>
+ <default-last-value-queue>false</default-last-value-queue>
+ <default-non-destructive>false</default-non-destructive>
<default-exclusive-queue>false</default-exclusive-queue>
<default-consumers-before-dispatch>0</default-consumers-before-dispatch>
<default-delay-before-dispatch>-1</default-delay-before-dispatch>
diff --git a/docs/user-manual/en/configuration-index.md b/docs/user-manual/en/configuration-index.md
index 9239123..d244b76 100644
--- a/docs/user-manual/en/configuration-index.md
+++ b/docs/user-manual/en/configuration-index.md
@@ -218,7 +218,9 @@ Name | Description | Default
[message-counter-history-day-limit](address-model.md) | Days to keep message counter data | 0
[last-value-queue](last-value-queues.md) | **deprecated** Queue is a last value queue; see `default-last-value-queue` instead | `false`
[default-last-value-queue](last-value-queues.md)| `last-value` value if none is set on the queue | `false`
+[default-last-value-key](last-value-queues.md)| `last-value-key` value if none is set on the queue | `null`
[default-exclusive-queue](exclusive-queues.md) | `exclusive` value if none is set on the queue | `false`
+[default-non-destructive](exclusive-queues.md) | `non-destructive` value if none is set on the queue | `false`
[default-consumers-before-dispatch](exclusive-queues.md) | `consumers-before-dispatch` value if none is set on the queue | 0
[default-delay-before-dispatch](exclusive-queues.md) | `delay-before-dispatch` value if none is set on the queue | -1
[redistribution-delay](clusters.md) | Timeout before redistributing values after no consumers | -1
@@ -234,6 +236,7 @@ Name | Description | Default
[auto-delete-queues](address-model.md#configuring-addresses-and-queues-via-address-settings) | Delete auto-created queues automatically | `true`
[auto-delete-created-queues](address-model.md#configuring-addresses-and-queues-via-address-settings) | Delete created queues automatically | `false`
[auto-delete-queues-delay](address-model.md#configuring-addresses-and-queues-via-address-settings) | Delay for deleting auto-created queues | 0
+[auto-delete-queues-message-count](address-model.md#configuring-addresses-and-queues-via-address-settings) | Message count the queue must be at or below before it can be auto deleted | 0
[config-delete-queues](config-reload.md)| How to deal with queues deleted from XML at runtime| `OFF`
[auto-create-addresses](address-model.md#configuring-addresses-and-queues-via-address-settings) | Create addresses automatically | `true`
[auto-delete-addresses](address-model.md#configuring-addresses-and-queues-via-address-settings) | Delete auto-created addresses automatically | `true`
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 6ba521f..8e52cbc 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
@@ -69,6 +69,7 @@ import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.ServerSession;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.settings.impl.DeletionPolicy;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.apache.activemq.artemis.core.transaction.impl.XidImpl;
import org.apache.activemq.artemis.nativo.jlibaio.LibaioContext;
@@ -721,23 +722,144 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
String addressFullMessagePolicy = "PAGE";
long slowConsumerThreshold = 5;
long slowConsumerCheckPeriod = 10;
- String slowConsumerPolicy = SlowConsumerPolicy.KILL.toString();
- boolean autoCreateJmsQueues = false;
- boolean autoDeleteJmsQueues = false;
- boolean autoCreateJmsTopics = false;
- boolean autoDeleteJmsTopics = false;
-
- serverControl.addAddressSettings(addressMatch, DLA, expiryAddress, expiryDelay, lastValueQueue, deliveryAttempts, maxSizeBytes, pageSizeBytes, pageMaxCacheSize, redeliveryDelay, redeliveryMultiplier, maxRedeliveryDelay, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy, slowConsumerThreshold, slowConsumerCheckPeriod, slowConsumerPolicy, autoCreateJmsQueues, autoDeleteJmsQueues, autoCreateJmsTopics, autoDeleteJmsTopics);
+ String slowConsumerPolicy = SlowConsumerPolicy.getType(RandomUtil.randomPositiveInt() % 2).toString();
+ boolean autoCreateJmsQueues = RandomUtil.randomBoolean();
+ boolean autoDeleteJmsQueues = RandomUtil.randomBoolean();
+ boolean autoCreateJmsTopics = RandomUtil.randomBoolean();
+ boolean autoDeleteJmsTopics = RandomUtil.randomBoolean();
+
+ boolean autoCreateQueues = RandomUtil.randomBoolean();
+ boolean autoDeleteQueues = RandomUtil.randomBoolean();
+ boolean autoCreateAddresses = RandomUtil.randomBoolean();
+ boolean autoDeleteAddresses = RandomUtil.randomBoolean();
+ String configDeleteQueues = DeletionPolicy.getType(RandomUtil.randomPositiveInt() % 2).toString();
+ String configDeleteAddresses = DeletionPolicy.getType(RandomUtil.randomPositiveInt() % 2).toString();
+ long maxSizeBytesRejectThreshold = RandomUtil.randomPositiveLong();
+ String defaultLastValueKey = RandomUtil.randomString();
+ boolean defaultNonDestructive = RandomUtil.randomBoolean();
+ boolean defaultExclusiveQueue = RandomUtil.randomBoolean();
+ boolean defaultGroupRebalance = RandomUtil.randomBoolean();
+ int defaultGroupBuckets = RandomUtil.randomPositiveInt();
+ String defaultGroupFirstKey = RandomUtil.randomString();
+ int defaultMaxConsumers = RandomUtil.randomPositiveInt();
+ boolean defaultPurgeOnNoConsumers = RandomUtil.randomBoolean();
+ int defaultConsumersBeforeDispatch = RandomUtil.randomPositiveInt();
+ long defaultDelayBeforeDispatch = RandomUtil.randomPositiveLong();
+ String defaultQueueRoutingType = RoutingType.getType((byte) (RandomUtil.randomPositiveInt() % 2)).toString();
+ String defaultAddressRoutingType = RoutingType.getType((byte) (RandomUtil.randomPositiveInt() % 2)).toString();
+ int defaultConsumerWindowSize = RandomUtil.randomPositiveInt();
+ long defaultRingSize = RandomUtil.randomPositiveLong();
+ boolean autoDeleteCreatedQueues = RandomUtil.randomBoolean();
+ long autoDeleteQueuesDelay = RandomUtil.randomPositiveLong();
+ long autoDeleteQueuesMessageCount = RandomUtil.randomPositiveLong();
+ long autoDeleteAddressesDelay = RandomUtil.randomPositiveLong();
+ double redeliveryCollisionAvoidanceFactor = RandomUtil.randomDouble();
+
+ serverControl.addAddressSettings(addressMatch,
+ DLA,
+ expiryAddress,
+ expiryDelay,
+ lastValueQueue,
+ deliveryAttempts,
+ maxSizeBytes,
+ pageSizeBytes,
+ pageMaxCacheSize,
+ redeliveryDelay,
+ redeliveryMultiplier,
+ maxRedeliveryDelay,
+ redistributionDelay,
+ sendToDLAOnNoRoute,
+ addressFullMessagePolicy,
+ slowConsumerThreshold,
+ slowConsumerCheckPeriod,
+ slowConsumerPolicy,
+ autoCreateJmsQueues,
+ autoDeleteJmsQueues,
+ autoCreateJmsTopics,
+ autoDeleteJmsTopics,
+ autoCreateQueues,
+ autoDeleteQueues,
+ autoCreateAddresses,
+ autoDeleteAddresses,
+ configDeleteQueues,
+ configDeleteAddresses,
+ maxSizeBytesRejectThreshold,
+ defaultLastValueKey,
+ defaultNonDestructive,
+ defaultExclusiveQueue,
+ defaultGroupRebalance,
+ defaultGroupBuckets,
+ defaultGroupFirstKey,
+ defaultMaxConsumers,
+ defaultPurgeOnNoConsumers,
+ defaultConsumersBeforeDispatch,
+ defaultDelayBeforeDispatch,
+ defaultQueueRoutingType,
+ defaultAddressRoutingType,
+ defaultConsumerWindowSize,
+ defaultRingSize,
+ autoDeleteCreatedQueues,
+ autoDeleteQueuesDelay,
+ autoDeleteQueuesMessageCount,
+ autoDeleteAddressesDelay,
+ redeliveryCollisionAvoidanceFactor);
boolean ex = false;
try {
- serverControl.addAddressSettings(addressMatch, DLA, expiryAddress, expiryDelay, lastValueQueue, deliveryAttempts, 100, 1000, pageMaxCacheSize, redeliveryDelay, redeliveryMultiplier, maxRedeliveryDelay, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy, slowConsumerThreshold, slowConsumerCheckPeriod, slowConsumerPolicy, autoCreateJmsQueues, autoDeleteJmsQueues, autoCreateJmsTopics, autoDeleteJmsTopics);
+ serverControl.addAddressSettings(addressMatch,
+ DLA,
+ expiryAddress,
+ expiryDelay,
+ lastValueQueue,
+ deliveryAttempts,
+ 100,
+ 1000,
+ pageMaxCacheSize,
+ redeliveryDelay,
+ redeliveryMultiplier,
+ maxRedeliveryDelay,
+ redistributionDelay,
+ sendToDLAOnNoRoute,
+ addressFullMessagePolicy,
+ slowConsumerThreshold,
+ slowConsumerCheckPeriod,
+ slowConsumerPolicy,
+ autoCreateJmsQueues,
+ autoDeleteJmsQueues,
+ autoCreateJmsTopics,
+ autoDeleteJmsTopics,
+ autoCreateQueues,
+ autoDeleteQueues,
+ autoCreateAddresses,
+ autoDeleteAddresses,
+ configDeleteQueues,
+ configDeleteAddresses,
+ maxSizeBytesRejectThreshold,
+ defaultLastValueKey,
+ defaultNonDestructive,
+ defaultExclusiveQueue,
+ defaultGroupRebalance,
+ defaultGroupBuckets,
+ defaultGroupFirstKey,
+ defaultMaxConsumers,
+ defaultPurgeOnNoConsumers,
+ defaultConsumersBeforeDispatch,
+ defaultDelayBeforeDispatch,
+ defaultQueueRoutingType,
+ defaultAddressRoutingType,
+ defaultConsumerWindowSize,
+ defaultRingSize,
+ autoDeleteCreatedQueues,
+ autoDeleteQueuesDelay,
+ autoDeleteQueuesMessageCount,
+ autoDeleteAddressesDelay,
+ redeliveryCollisionAvoidanceFactor);
} catch (Exception expected) {
ex = true;
}
assertTrue("Exception expected", ex);
- //restartServer();
+ restartServer();
serverControl = createManagementControl();
String jsonString = serverControl.getAddressSettingsAsJSON(exactAddress);
@@ -761,10 +883,83 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
assertEquals(slowConsumerPolicy, info.getSlowConsumerPolicy());
assertEquals(autoCreateJmsQueues, info.isAutoCreateJmsQueues());
assertEquals(autoDeleteJmsQueues, info.isAutoDeleteJmsQueues());
- // assertEquals(autoCreateJmsTopics, info.isAutoCreateAddresses());
+ assertEquals(autoCreateJmsTopics, info.isAutoCreateJmsTopics());
assertEquals(autoDeleteJmsTopics, info.isAutoDeleteJmsTopics());
-
- serverControl.addAddressSettings(addressMatch, DLA, expiryAddress, expiryDelay, lastValueQueue, deliveryAttempts, -1, 1000, pageMaxCacheSize, redeliveryDelay, redeliveryMultiplier, maxRedeliveryDelay, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy, slowConsumerThreshold, slowConsumerCheckPeriod, slowConsumerPolicy, autoCreateJmsQueues, autoDeleteJmsQueues, autoCreateJmsTopics, autoDeleteJmsTopics);
+ assertEquals(autoCreateQueues, info.isAutoCreateQueues());
+ assertEquals(autoDeleteQueues, info.isAutoDeleteQueues());
+ assertEquals(autoCreateAddresses, info.isAutoCreateAddresses());
+ assertEquals(autoDeleteAddresses, info.isAutoDeleteAddresses());
+ assertEquals(configDeleteQueues, info.getConfigDeleteQueues());
+ assertEquals(configDeleteAddresses, info.getConfigDeleteAddresses());
+ assertEquals(maxSizeBytesRejectThreshold, info.getMaxSizeBytesRejectThreshold());
+ assertEquals(defaultLastValueKey, info.getDefaultLastValueKey());
+ assertEquals(defaultNonDestructive, info.isDefaultNonDestructive());
+ assertEquals(defaultExclusiveQueue, info.isDefaultExclusiveQueue());
+ assertEquals(defaultGroupRebalance, info.isDefaultGroupRebalance());
+ assertEquals(defaultGroupBuckets, info.getDefaultGroupBuckets());
+ assertEquals(defaultGroupFirstKey, info.getDefaultGroupFirstKey());
+ assertEquals(defaultMaxConsumers, info.getDefaultMaxConsumers());
+ assertEquals(defaultPurgeOnNoConsumers, info.isDefaultPurgeOnNoConsumers());
+ assertEquals(defaultConsumersBeforeDispatch, info.getDefaultConsumersBeforeDispatch());
+ assertEquals(defaultDelayBeforeDispatch, info.getDefaultDelayBeforeDispatch());
+ assertEquals(defaultQueueRoutingType, info.getDefaultQueueRoutingType());
+ assertEquals(defaultAddressRoutingType, info.getDefaultAddressRoutingType());
+ assertEquals(defaultConsumerWindowSize, info.getDefaultConsumerWindowSize());
+ assertEquals(defaultRingSize, info.getDefaultRingSize());
+ assertEquals(autoDeleteCreatedQueues, info.isAutoDeleteCreatedQueues());
+ assertEquals(autoDeleteQueuesDelay, info.getAutoDeleteQueuesDelay());
+ assertEquals(autoDeleteQueuesMessageCount, info.getAutoDeleteQueuesMessageCount());
+ assertEquals(autoDeleteAddressesDelay, info.getAutoDeleteAddressesDelay());
+ assertEquals(redeliveryCollisionAvoidanceFactor, info.getRedeliveryCollisionAvoidanceFactor(), 0);
+
+ serverControl.addAddressSettings(addressMatch,
+ DLA,
+ expiryAddress,
+ expiryDelay,
+ lastValueQueue,
+ deliveryAttempts,
+ -1,
+ 1000,
+ pageMaxCacheSize,
+ redeliveryDelay,
+ redeliveryMultiplier,
+ maxRedeliveryDelay,
+ redistributionDelay,
+ sendToDLAOnNoRoute,
+ addressFullMessagePolicy,
+ slowConsumerThreshold,
+ slowConsumerCheckPeriod,
+ slowConsumerPolicy,
+ autoCreateJmsQueues,
+ autoDeleteJmsQueues,
+ autoCreateJmsTopics,
+ autoDeleteJmsTopics,
+ autoCreateQueues,
+ autoDeleteQueues,
+ autoCreateAddresses,
+ autoDeleteAddresses,
+ configDeleteQueues,
+ configDeleteAddresses,
+ maxSizeBytesRejectThreshold,
+ defaultLastValueKey,
+ defaultNonDestructive,
+ defaultExclusiveQueue,
+ defaultGroupRebalance,
+ defaultGroupBuckets,
+ defaultGroupFirstKey,
+ defaultMaxConsumers,
+ defaultPurgeOnNoConsumers,
+ defaultConsumersBeforeDispatch,
+ defaultDelayBeforeDispatch,
+ defaultQueueRoutingType,
+ defaultAddressRoutingType,
+ defaultConsumerWindowSize,
+ defaultRingSize,
+ autoDeleteCreatedQueues,
+ autoDeleteQueuesDelay,
+ autoDeleteQueuesMessageCount,
+ autoDeleteAddressesDelay,
+ redeliveryCollisionAvoidanceFactor);
jsonString = serverControl.getAddressSettingsAsJSON(exactAddress);
info = AddressSettingsInfo.from(jsonString);
@@ -787,12 +982,85 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
assertEquals(slowConsumerPolicy, info.getSlowConsumerPolicy());
assertEquals(autoCreateJmsQueues, info.isAutoCreateJmsQueues());
assertEquals(autoDeleteJmsQueues, info.isAutoDeleteJmsQueues());
- // assertEquals(autoCreateJmsTopics, info.isAutoCreateAddresses());
+ assertEquals(autoCreateJmsTopics, info.isAutoCreateJmsTopics());
assertEquals(autoDeleteJmsTopics, info.isAutoDeleteJmsTopics());
+ assertEquals(autoCreateQueues, info.isAutoCreateQueues());
+ assertEquals(autoDeleteQueues, info.isAutoDeleteQueues());
+ assertEquals(autoCreateAddresses, info.isAutoCreateAddresses());
+ assertEquals(autoDeleteAddresses, info.isAutoDeleteAddresses());
+ assertEquals(configDeleteQueues, info.getConfigDeleteQueues());
+ assertEquals(configDeleteAddresses, info.getConfigDeleteAddresses());
+ assertEquals(maxSizeBytesRejectThreshold, info.getMaxSizeBytesRejectThreshold());
+ assertEquals(defaultLastValueKey, info.getDefaultLastValueKey());
+ assertEquals(defaultNonDestructive, info.isDefaultNonDestructive());
+ assertEquals(defaultExclusiveQueue, info.isDefaultExclusiveQueue());
+ assertEquals(defaultGroupRebalance, info.isDefaultGroupRebalance());
+ assertEquals(defaultGroupBuckets, info.getDefaultGroupBuckets());
+ assertEquals(defaultGroupFirstKey, info.getDefaultGroupFirstKey());
+ assertEquals(defaultMaxConsumers, info.getDefaultMaxConsumers());
+ assertEquals(defaultPurgeOnNoConsumers, info.isDefaultPurgeOnNoConsumers());
+ assertEquals(defaultConsumersBeforeDispatch, info.getDefaultConsumersBeforeDispatch());
+ assertEquals(defaultDelayBeforeDispatch, info.getDefaultDelayBeforeDispatch());
+ assertEquals(defaultQueueRoutingType, info.getDefaultQueueRoutingType());
+ assertEquals(defaultAddressRoutingType, info.getDefaultAddressRoutingType());
+ assertEquals(defaultConsumerWindowSize, info.getDefaultConsumerWindowSize());
+ assertEquals(defaultRingSize, info.getDefaultRingSize());
+ assertEquals(autoDeleteCreatedQueues, info.isAutoDeleteCreatedQueues());
+ assertEquals(autoDeleteQueuesDelay, info.getAutoDeleteQueuesDelay());
+ assertEquals(autoDeleteQueuesMessageCount, info.getAutoDeleteQueuesMessageCount());
+ assertEquals(autoDeleteAddressesDelay, info.getAutoDeleteAddressesDelay());
+ assertEquals(redeliveryCollisionAvoidanceFactor, info.getRedeliveryCollisionAvoidanceFactor(), 0);
ex = false;
try {
- serverControl.addAddressSettings(addressMatch, DLA, expiryAddress, expiryDelay, lastValueQueue, deliveryAttempts, -2, 1000, pageMaxCacheSize, redeliveryDelay, redeliveryMultiplier, maxRedeliveryDelay, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy, slowConsumerThreshold, slowConsumerCheckPeriod, slowConsumerPolicy, autoCreateJmsQueues, autoDeleteJmsQueues, autoCreateJmsTopics, autoDeleteJmsTopics);
+ serverControl.addAddressSettings(addressMatch,
+ DLA,
+ expiryAddress,
+ expiryDelay,
+ lastValueQueue,
+ deliveryAttempts,
+ -2,
+ 1000,
+ pageMaxCacheSize,
+ redeliveryDelay,
+ redeliveryMultiplier,
+ maxRedeliveryDelay,
+ redistributionDelay,
+ sendToDLAOnNoRoute,
+ addressFullMessagePolicy,
+ slowConsumerThreshold,
+ slowConsumerCheckPeriod,
+ slowConsumerPolicy,
+ autoCreateJmsQueues,
+ autoDeleteJmsQueues,
+ autoCreateJmsTopics,
+ autoDeleteJmsTopics,
+ autoCreateQueues,
+ autoDeleteQueues,
+ autoCreateAddresses,
+ autoDeleteAddresses,
+ configDeleteQueues,
+ configDeleteAddresses,
+ maxSizeBytesRejectThreshold,
+ defaultLastValueKey,
+ defaultNonDestructive,
+ defaultExclusiveQueue,
+ defaultGroupRebalance,
+ defaultGroupBuckets,
+ defaultGroupFirstKey,
+ defaultMaxConsumers,
+ defaultPurgeOnNoConsumers,
+ defaultConsumersBeforeDispatch,
+ defaultDelayBeforeDispatch,
+ defaultQueueRoutingType,
+ defaultAddressRoutingType,
+ defaultConsumerWindowSize,
+ defaultRingSize,
+ autoDeleteCreatedQueues,
+ autoDeleteQueuesDelay,
+ autoDeleteQueuesMessageCount,
+ autoDeleteAddressesDelay,
+ redeliveryCollisionAvoidanceFactor);
} catch (Exception e) {
ex = true;
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java
index 7278114..c727b07 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java
@@ -22,7 +22,6 @@ import org.apache.activemq.artemis.api.core.ActiveMQAddressDoesNotExistException
import org.apache.activemq.artemis.api.core.management.ActiveMQServerControl;
import org.apache.activemq.artemis.api.core.management.Parameter;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
-import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTest {
@@ -826,7 +825,29 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
@Parameter(desc = "allow auto-created queues to be deleted automatically", name = "autoDeleteJmsQueues") boolean autoDeleteJmsQueues,
@Parameter(desc = "allow topics to be created automatically", name = "autoCreateJmsTopics") boolean autoCreateJmsTopics,
@Parameter(desc = "allow auto-created topics to be deleted automatically", name = "autoDeleteJmsTopics") boolean autoDeleteJmsTopics) throws Exception {
- addAddressSettings(addressMatch, DLA, expiryAddress, expiryDelay, lastValueQueue, deliveryAttempts, maxSizeBytes, pageSizeBytes, pageMaxCacheSize, redeliveryDelay, redeliveryMultiplier, maxRedeliveryDelay, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy, slowConsumerThreshold, slowConsumerCheckPeriod, slowConsumerPolicy, autoCreateJmsQueues, autoDeleteJmsQueues, autoCreateJmsTopics, autoDeleteJmsTopics, AddressSettings.DEFAULT_AUTO_CREATE_QUEUES, AddressSett [...]
+ proxy.invokeOperation("addAddressSettings",
+ addressMatch,
+ DLA,
+ expiryAddress,
+ expiryDelay,
+ lastValueQueue,
+ deliveryAttempts,
+ maxSizeBytes,
+ pageSizeBytes,
+ pageMaxCacheSize,
+ redeliveryDelay,
+ redeliveryMultiplier,
+ maxRedeliveryDelay,
+ redistributionDelay,
+ sendToDLAOnNoRoute,
+ addressFullMessagePolicy,
+ slowConsumerThreshold,
+ slowConsumerCheckPeriod,
+ slowConsumerPolicy,
+ autoCreateJmsQueues,
+ autoDeleteJmsQueues,
+ autoCreateJmsTopics,
+ autoDeleteJmsTopics);
}
@Override
@@ -856,7 +877,133 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
@Parameter(desc = "allow auto-created queues to be deleted automatically", name = "autoDeleteQueues") boolean autoDeleteQueues,
@Parameter(desc = "allow topics to be created automatically", name = "autoCreateAddresses") boolean autoCreateAddresses,
@Parameter(desc = "allow auto-created topics to be deleted automatically", name = "autoDeleteAddresses") boolean autoDeleteAddresses) throws Exception {
- proxy.invokeOperation("addAddressSettings", addressMatch, DLA, expiryAddress, expiryDelay, lastValueQueue, deliveryAttempts, maxSizeBytes, pageSizeBytes, pageMaxCacheSize, redeliveryDelay, redeliveryMultiplier, maxRedeliveryDelay, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy, slowConsumerThreshold, slowConsumerCheckPeriod, slowConsumerPolicy, autoCreateJmsQueues, autoDeleteJmsQueues, autoCreateJmsTopics, autoDeleteJmsTopics, autoCreateQueues, autoDeleteQu [...]
+ proxy.invokeOperation("addAddressSettings",
+ addressMatch,
+ DLA,
+ expiryAddress,
+ expiryDelay,
+ lastValueQueue,
+ deliveryAttempts,
+ maxSizeBytes,
+ pageSizeBytes,
+ pageMaxCacheSize,
+ redeliveryDelay,
+ redeliveryMultiplier,
+ maxRedeliveryDelay,
+ redistributionDelay,
+ sendToDLAOnNoRoute,
+ addressFullMessagePolicy,
+ slowConsumerThreshold,
+ slowConsumerCheckPeriod,
+ slowConsumerPolicy,
+ autoCreateJmsQueues,
+ autoDeleteJmsQueues,
+ autoCreateJmsTopics,
+ autoDeleteJmsTopics,
+ autoCreateQueues,
+ autoDeleteQueues,
+ autoCreateAddresses,
+ autoDeleteAddresses);
+ }
+
+ @Override
+ public void addAddressSettings(@Parameter(desc = "an address match", name = "addressMatch") String addressMatch,
+ @Parameter(desc = "the dead letter address setting", name = "DLA") String DLA,
+ @Parameter(desc = "the expiry address setting", name = "expiryAddress") String expiryAddress,
+ @Parameter(desc = "the expiry delay setting", name = "expiryDelay") long expiryDelay,
+ @Parameter(desc = "are any queues created for this address a last value queue", name = "lastValueQueue") boolean lastValueQueue,
+ @Parameter(desc = "the delivery attempts", name = "deliveryAttempts") int deliveryAttempts,
+ @Parameter(desc = "the max size in bytes", name = "maxSizeBytes") long maxSizeBytes,
+ @Parameter(desc = "the page size in bytes", name = "pageSizeBytes") int pageSizeBytes,
+ @Parameter(desc = "the max number of pages in the soft memory cache", name = "pageMaxCacheSize") int pageMaxCacheSize,
+ @Parameter(desc = "the redelivery delay", name = "redeliveryDelay") long redeliveryDelay,
+ @Parameter(desc = "the redelivery delay multiplier", name = "redeliveryMultiplier") double redeliveryMultiplier,
+ @Parameter(desc = "the maximum redelivery delay", name = "maxRedeliveryDelay") long maxRedeliveryDelay,
+ @Parameter(desc = "the redistribution delay", name = "redistributionDelay") long redistributionDelay,
+ @Parameter(desc = "do we send to the DLA when there is no where to route the message", name = "sendToDLAOnNoRoute") boolean sendToDLAOnNoRoute,
+ @Parameter(desc = "the policy to use when the address is full", name = "addressFullMessagePolicy") String addressFullMessagePolicy,
+ @Parameter(desc = "when a consumer falls below this threshold in terms of messages consumed per second it will be considered 'slow'", name = "slowConsumerThreshold") long slowConsumerThreshold,
+ @Parameter(desc = "how often (in seconds) to check for slow consumers", name = "slowConsumerCheckPeriod") long slowConsumerCheckPeriod,
+ @Parameter(desc = "the policy to use when a slow consumer is detected", name = "slowConsumerPolicy") String slowConsumerPolicy,
+ @Parameter(desc = "allow jms queues to be created automatically", name = "autoCreateJmsQueues") boolean autoCreateJmsQueues,
+ @Parameter(desc = "allow auto-created jms queues to be deleted automatically", name = "autoDeleteJmsQueues") boolean autoDeleteJmsQueues,
+ @Parameter(desc = "allow jms topics to be created automatically", name = "autoCreateJmsTopics") boolean autoCreateJmsTopics,
+ @Parameter(desc = "allow auto-created jms topics to be deleted automatically", name = "autoDeleteJmsTopics") boolean autoDeleteJmsTopics,
+ @Parameter(desc = "allow queues to be created automatically", name = "autoCreateQueues") boolean autoCreateQueues,
+ @Parameter(desc = "allow auto-created queues to be deleted automatically", name = "autoDeleteQueues") boolean autoDeleteQueues,
+ @Parameter(desc = "allow topics to be created automatically", name = "autoCreateAddresses") boolean autoCreateAddresses,
+ @Parameter(desc = "allow auto-created topics to be deleted automatically", name = "autoDeleteAddresses") boolean autoDeleteAddresses,
+ @Parameter(desc = "how to deal with queues deleted from XML at runtime", name = "configDeleteQueues") String configDeleteQueues,
+ @Parameter(desc = "how to deal with addresses deleted from XML at runtime", name = "configDeleteAddresses") String configDeleteAddresses,
+ @Parameter(desc = "used with `BLOCK`, the max size an address can reach before messages are rejected; works in combination with `max-size-bytes` for AMQP clients only", name = "maxSizeBytesRejectThreshold") long maxSizeBytesRejectThreshold,
+ @Parameter(desc = "last-value-key value if none is set on the queue", name = "defaultLastValueKey") String defaultLastValueKey,
+ @Parameter(desc = "non-destructive value if none is set on the queue", name = "defaultNonDestructive") boolean defaultNonDestructive,
+ @Parameter(desc = "exclusive value if none is set on the queue", name = "defaultExclusiveQueue") boolean defaultExclusiveQueue,
+ @Parameter(desc = "group-rebalance value if none is set on the queue", name = "defaultGroupRebalance") boolean defaultGroupRebalance,
+ @Parameter(desc = "group-buckets value if none is set on the queue", name = "defaultGroupBuckets") int defaultGroupBuckets,
+ @Parameter(desc = "group-first-key value if none is set on the queue", name = "defaultGroupFirstKey") String defaultGroupFirstKey,
+ @Parameter(desc = "max-consumers value if none is set on the queue", name = "defaultMaxConsumers") int defaultMaxConsumers,
+ @Parameter(desc = "purge-on-no-consumers value if none is set on the queue", name = "defaultPurgeOnNoConsumers") boolean defaultPurgeOnNoConsumers,
+ @Parameter(desc = "consumers-before-dispatch value if none is set on the queue", name = "defaultConsumersBeforeDispatch") int defaultConsumersBeforeDispatch,
+ @Parameter(desc = "delay-before-dispatch value if none is set on the queue", name = "defaultDelayBeforeDispatch") long defaultDelayBeforeDispatch,
+ @Parameter(desc = "routing-type value if none is set on the queue", name = "defaultQueueRoutingType") String defaultQueueRoutingType,
+ @Parameter(desc = "routing-type value if none is set on the address", name = "defaultAddressRoutingType") String defaultAddressRoutingType,
+ @Parameter(desc = "consumer-window-size value if none is set on the queue", name = "defaultConsumerWindowSize") int defaultConsumerWindowSize,
+ @Parameter(desc = "ring-size value if none is set on the queue", name = "defaultRingSize") long defaultRingSize,
+ @Parameter(desc = "allow created queues to be deleted automatically", name = "autoDeleteCreatedQueues") boolean autoDeleteCreatedQueues,
+ @Parameter(desc = "delay for deleting auto-created queues", name = "autoDeleteQueuesDelay") long autoDeleteQueuesDelay,
+ @Parameter(desc = "the message count the queue must be at or below before it can be auto deleted", name = "autoDeleteQueuesMessageCount") long autoDeleteQueuesMessageCount,
+ @Parameter(desc = "delay for deleting auto-created addresses", name = "autoDeleteAddressesDelay") long autoDeleteAddressesDelay,
+ @Parameter(desc = "factor by which to modify the redelivery delay slightly to avoid collisions", name = "redeliveryCollisionAvoidanceFactor") double redeliveryCollisionAvoidanceFactor) throws Exception {
+ proxy.invokeOperation("addAddressSettings",
+ addressMatch,
+ DLA,
+ expiryAddress,
+ expiryDelay,
+ lastValueQueue,
+ deliveryAttempts,
+ maxSizeBytes,
+ pageSizeBytes,
+ pageMaxCacheSize,
+ redeliveryDelay,
+ redeliveryMultiplier,
+ maxRedeliveryDelay,
+ redistributionDelay,
+ sendToDLAOnNoRoute,
+ addressFullMessagePolicy,
+ slowConsumerThreshold,
+ slowConsumerCheckPeriod,
+ slowConsumerPolicy,
+ autoCreateJmsQueues,
+ autoDeleteJmsQueues,
+ autoCreateJmsTopics,
+ autoDeleteJmsTopics,
+ autoCreateQueues,
+ autoDeleteQueues,
+ autoCreateAddresses,
+ autoDeleteAddresses,
+ configDeleteQueues,
+ configDeleteAddresses,
+ maxSizeBytesRejectThreshold,
+ defaultLastValueKey,
+ defaultNonDestructive,
+ defaultExclusiveQueue,
+ defaultGroupRebalance,
+ defaultGroupBuckets,
+ defaultGroupFirstKey,
+ defaultMaxConsumers,
+ defaultPurgeOnNoConsumers,
+ defaultConsumersBeforeDispatch,
+ defaultDelayBeforeDispatch,
+ defaultQueueRoutingType,
+ defaultAddressRoutingType,
+ defaultConsumerWindowSize,
+ defaultRingSize,
+ autoDeleteCreatedQueues,
+ autoDeleteQueuesDelay,
+ autoDeleteQueuesMessageCount,
+ autoDeleteAddressesDelay,
+ redeliveryCollisionAvoidanceFactor);
}
@Override
[activemq-artemis] 02/03: ARTEMIS-2504 implement retroactive
addresses
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 84067d8fef311d641ef7d35b9cdda3bc48759a37
Author: Justin Bertram <jb...@apache.org>
AuthorDate: Wed Aug 28 10:27:52 2019 -0500
ARTEMIS-2504 implement retroactive addresses
A new feature to preserve messages sent to an address for queues that will be
created on the address in the future. This is essentially equivalent to the
"retroactive consumer" feature from 5.x. However, it's implemented in a way
that fits with the address model of Artemis.
---
.../apache/activemq/artemis/logs/AuditLogger.java | 7 +
.../api/config/ActiveMQDefaultConfiguration.java | 6 +
.../api/core/management/ActiveMQServerControl.java | 3 +-
.../api/core/management/AddressControl.java | 3 +
.../api/core/management/AddressSettingsInfo.java | 13 +-
.../artemis/api/core/management/DivertControl.java | 3 +
.../artemis/api/core/management/QueueControl.java | 6 +
.../artemis/api/core/management/ResourceNames.java | 31 +-
.../core/management/AddressSettingsInfoTest.java | 4 +-
.../api/core/management/ResourceNamesTest.java | 91 ++++
.../deployers/impl/FileConfigurationParser.java | 6 +
.../management/impl/ActiveMQServerControlImpl.java | 14 +-
.../core/management/impl/AddressControlImpl.java | 8 +
.../core/management/impl/DivertControlImpl.java | 15 +-
.../core/management/impl/QueueControlImpl.java | 16 +
.../core/postoffice/impl/PostOfficeImpl.java | 133 ++++++
.../core/postoffice/impl/SimpleAddressManager.java | 2 +-
.../artemis/core/server/RoutingContext.java | 2 +-
.../core/server/impl/ActiveMQServerImpl.java | 20 +-
.../artemis/core/server/impl/AddressInfo.java | 14 +-
.../artemis/core/server/impl/DivertImpl.java | 2 +-
.../artemis/core/server/impl/JournalLoader.java | 3 +-
.../core/server/impl/PostOfficeJournalLoader.java | 5 +-
.../artemis/core/server/impl/QueueImpl.java | 33 +-
.../core/server/impl/RoutingContextImpl.java | 3 +-
.../management/impl/ManagementServiceImpl.java | 2 +-
.../core/settings/impl/AddressSettings.java | 32 +-
.../resources/schema/artemis-configuration.xsd | 8 +
.../core/config/impl/FileConfigurationTest.java | 2 +
.../impl/journal/AddressBindingEncodingTest.java | 52 +++
.../resources/ConfigurationTest-full-config.xml | 1 +
...rationTest-xinclude-config-address-settings.xml | 1 +
.../src/test/resources/artemis-configuration.xsd | 15 +
docs/user-manual/en/SUMMARY.md | 1 +
docs/user-manual/en/address-model.md | 10 +-
docs/user-manual/en/configuration-index.md | 1 +
docs/user-manual/en/retroactive-addresses.md | 88 ++++
.../RequestReplyMultiProtocolTest.java | 5 +-
.../crossprotocol/RequestReplyNonJMSTest.java | 14 +-
.../management/ActiveMQServerControlTest.java | 16 +-
.../ActiveMQServerControlUsingCoreTest.java | 6 +-
.../integration/management/AddressControlTest.java | 13 +
.../management/AddressControlUsingCoreTest.java | 5 +
.../integration/management/DivertControlTest.java | 24 ++
.../management/DivertControlUsingCoreTest.java | 5 +
.../integration/management/QueueControlTest.java | 24 +-
.../management/QueueControlUsingCoreTest.java | 5 +
.../server/RetroactiveAddressFailoverTest.java | 119 ++++++
.../integration/server/RetroactiveAddressTest.java | 476 +++++++++++++++++++++
.../core/server/impl/fakes/FakeJournalLoader.java | 3 +-
50 files changed, 1320 insertions(+), 51 deletions(-)
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/AuditLogger.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/AuditLogger.java
index cbea7b9..70abebd 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/AuditLogger.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/AuditLogger.java
@@ -2277,4 +2277,11 @@ public interface AuditLogger extends BasicLogger {
void getRingSize(String user, Object source, Object... args);
+ static void isRetroactiveResource(Object source) {
+ LOGGER.isRetroactiveResource(getCaller(), source);
+ }
+
+ @LogMessage(level = Logger.Level.INFO)
+ @Message(id = 601503, value = "User {0} is getting retroactiveResource property on target resource: {1} {2}", format = Message.Format.MESSAGE_FORMAT)
+ void isRetroactiveResource(String user, Object source, Object... args);
}
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 d41e8d3..76e1557 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
@@ -568,6 +568,8 @@ public final class ActiveMQDefaultConfiguration {
public static final long DEFAULT_ANALYZE_CRITICAL_TIMEOUT = 120000;
+ public static final long DEFAULT_RETROACTIVE_MESSAGE_COUNT = 0;
+
public static final CriticalAnalyzerPolicy DEFAULT_ANALYZE_CRITICAL_POLICY = CriticalAnalyzerPolicy.LOG;
/**
@@ -1437,6 +1439,10 @@ public final class ActiveMQDefaultConfiguration {
return DEFAULT_RING_SIZE;
}
+ public static long getDefaultRetroactiveMessageCount() {
+ return DEFAULT_RETROACTIVE_MESSAGE_COUNT;
+ }
+
public static int getDefaultConsumersBeforeDispatch() {
return DEFAULT_CONSUMERS_BEFORE_DISPATCH;
}
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java
index 207a8cc..04b267c 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java
@@ -1384,7 +1384,8 @@ public interface ActiveMQServerControl {
@Parameter(desc = "delay for deleting auto-created queues", name = "autoDeleteQueuesDelay") long autoDeleteQueuesDelay,
@Parameter(desc = "the message count the queue must be at or below before it can be auto deleted", name = "autoDeleteQueuesMessageCount") long autoDeleteQueuesMessageCount,
@Parameter(desc = "delay for deleting auto-created addresses", name = "autoDeleteAddressesDelay") long autoDeleteAddressesDelay,
- @Parameter(desc = "factor by which to modify the redelivery delay slightly to avoid collisions", name = "redeliveryCollisionAvoidanceFactor") double redeliveryCollisionAvoidanceFactor) throws Exception;
+ @Parameter(desc = "factor by which to modify the redelivery delay slightly to avoid collisions", name = "redeliveryCollisionAvoidanceFactor") double redeliveryCollisionAvoidanceFactor,
+ @Parameter(desc = "the number of messages to preserve for future queues created on the matching address", name = "retroactiveMessageCount") long retroactiveMessageCount) throws Exception;
@Operation(desc = "Remove address settings", impact = MBeanOperationInfo.ACTION)
void removeAddressSettings(@Parameter(desc = "an address match", name = "addressMatch") String addressMatch) throws Exception;
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressControl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressControl.java
index ce964aa..0eaf3e1 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressControl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressControl.java
@@ -162,4 +162,7 @@ public interface AddressControl {
@Attribute(desc = "indicates if the queues bound to this address are paused")
boolean isPaused();
+ @Attribute(desc = "whether this address is used for a retroactive address")
+ boolean isRetroactiveResource();
+
}
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfo.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfo.java
index 53b84d5..52854f2 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfo.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfo.java
@@ -115,6 +115,8 @@ public final class AddressSettingsInfo {
private final double redeliveryCollisionAvoidanceFactor;
+ private final long retroactiveMessageCount;
+
// Static --------------------------------------------------------
public static AddressSettingsInfo from(final String jsonString) {
@@ -164,7 +166,8 @@ public final class AddressSettingsInfo {
object.getJsonNumber("autoDeleteQueuesDelay").longValue(),
object.getJsonNumber("autoDeleteQueuesMessageCount").longValue(),
object.getJsonNumber("autoDeleteAddressesDelay").longValue(),
- object.getJsonNumber("redeliveryCollisionAvoidanceFactor").doubleValue());
+ object.getJsonNumber("redeliveryCollisionAvoidanceFactor").doubleValue(),
+ object.getJsonNumber("retroactiveMessageCount").longValue());
}
// Constructors --------------------------------------------------
@@ -214,7 +217,8 @@ public final class AddressSettingsInfo {
long autoDeleteQueuesDelay,
long autoDeleteQueuesMessageCount,
long autoDeleteAddressesDelay,
- double redeliveryCollisionAvoidanceFactor) {
+ double redeliveryCollisionAvoidanceFactor,
+ long retroactiveMessageCount) {
this.addressFullMessagePolicy = addressFullMessagePolicy;
this.maxSizeBytes = maxSizeBytes;
this.pageSizeBytes = pageSizeBytes;
@@ -261,6 +265,7 @@ public final class AddressSettingsInfo {
this.autoDeleteQueuesMessageCount = autoDeleteQueuesMessageCount;
this.autoDeleteAddressesDelay = autoDeleteAddressesDelay;
this.redeliveryCollisionAvoidanceFactor = redeliveryCollisionAvoidanceFactor;
+ this.retroactiveMessageCount = retroactiveMessageCount;
}
// Public --------------------------------------------------------
@@ -456,5 +461,9 @@ public final class AddressSettingsInfo {
public double getRedeliveryCollisionAvoidanceFactor() {
return redeliveryCollisionAvoidanceFactor;
}
+
+ public long getRetroactiveMessageCount() {
+ return retroactiveMessageCount;
+ }
}
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/DivertControl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/DivertControl.java
index 218ae50..fb15fe1 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/DivertControl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/DivertControl.java
@@ -85,4 +85,7 @@ public interface DivertControl {
*/
@Attribute(desc = "routing type used by this divert")
String getRoutingType();
+
+ @Attribute(desc = "whether this divert is for a retroactive address")
+ boolean isRetroactiveResource();
}
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 192815d..7942083 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
@@ -70,6 +70,12 @@ public interface QueueControl {
boolean isTemporary();
/**
+ * Returns whether this queue is used for a retroactive address.
+ */
+ @Attribute(desc = "whether this queue is used for a retroactive address")
+ boolean isRetroactiveResource();
+
+ /**
* Returns whether this queue is durable.
*/
@Attribute(desc = "whether this queue is durable")
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ResourceNames.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ResourceNames.java
index 377c234..410ee93 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ResourceNames.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ResourceNames.java
@@ -16,6 +16,9 @@
*/
package org.apache.activemq.artemis.api.core.management;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+
/**
* Helper class used to build resource names used by management messages.
* <br>
@@ -39,7 +42,33 @@ public final class ResourceNames {
public static final String BROADCAST_GROUP = "broadcastgroup.";
- private ResourceNames() {
+ public static final String RETROACTIVE_SUFFIX = "retro";
+
+ public static SimpleString getRetroactiveResourceQueueName(String prefix, String delimiter, SimpleString address, RoutingType routingType) {
+ return getRetroactiveResourceName(prefix, delimiter, address, trimLastCharacter(QUEUE).concat(delimiter).concat(routingType.toString().toLowerCase()));
+ }
+
+ public static SimpleString getRetroactiveResourceAddressName(String prefix, String delimiter, SimpleString address) {
+ return getRetroactiveResourceName(prefix, delimiter, address, trimLastCharacter(ADDRESS));
}
+ public static SimpleString getRetroactiveResourceDivertName(String prefix, String delimiter, SimpleString address) {
+ return getRetroactiveResourceName(prefix, delimiter, address, trimLastCharacter(DIVERT));
+ }
+
+ private static SimpleString getRetroactiveResourceName(String prefix, String delimiter, SimpleString address, String resourceType) {
+ return SimpleString.toSimpleString(prefix.concat(address.toString()).concat(delimiter).concat(resourceType).concat(delimiter).concat(RETROACTIVE_SUFFIX));
+ }
+
+ public static boolean isRetroactiveResource(String prefix, SimpleString address) {
+ return address.toString().startsWith(prefix) && address.toString().endsWith(RETROACTIVE_SUFFIX);
+ }
+
+ public static String decomposeRetroactiveResourceAddressName(String prefix, String delimiter, String address) {
+ return address.substring(address.indexOf(prefix) + prefix.length(), address.indexOf(delimiter + trimLastCharacter(ADDRESS)));
+ }
+
+ private static String trimLastCharacter(String toTrim) {
+ return toTrim.substring(0, toTrim.length() - 1);
+ }
}
diff --git a/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfoTest.java b/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfoTest.java
index 65f16fa..06a1d34 100644
--- a/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfoTest.java
+++ b/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/AddressSettingsInfoTest.java
@@ -76,7 +76,8 @@ public class AddressSettingsInfoTest {
"\"autoDeleteQueuesDelay\":4,\n" +
"\"autoDeleteQueuesMessageCount\":8,\n" +
"\"autoDeleteAddressesDelay\":3003,\n" +
- "\"redeliveryCollisionAvoidanceFactor\":1.1\n" +
+ "\"redeliveryCollisionAvoidanceFactor\":1.1,\n" +
+ "\"retroactiveMessageCount\":101\n" +
"}";
AddressSettingsInfo addressSettingsInfo = AddressSettingsInfo.from(json);
assertEquals("fullPolicy", addressSettingsInfo.getAddressFullMessagePolicy());
@@ -125,6 +126,7 @@ public class AddressSettingsInfoTest {
assertEquals(8, addressSettingsInfo.getAutoDeleteQueuesMessageCount());
assertEquals(3003, addressSettingsInfo.getAutoDeleteAddressesDelay());
assertEquals(1.1, addressSettingsInfo.getRedeliveryCollisionAvoidanceFactor(), 0);
+ assertEquals(101, addressSettingsInfo.getRetroactiveMessageCount());
}
}
diff --git a/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/ResourceNamesTest.java b/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/ResourceNamesTest.java
new file mode 100644
index 0000000..618f7db
--- /dev/null
+++ b/artemis-core-client/src/test/java/org/apache/activemq/artemis/api/core/management/ResourceNamesTest.java
@@ -0,0 +1,91 @@
+/**
+ * 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.api.core.management;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.UUID;
+
+import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ResourceNamesTest extends Assert {
+
+ char delimiterChar;
+ final String delimiter;
+ final SimpleString testAddress;
+ final String prefix;
+ final String baseName;
+ final String testResourceAddressName;
+ final String testResourceMulticastQueueName;
+ final String testResourceAnycastQueueName;
+ final String testResourceDivertName;
+
+
+ @Parameterized.Parameters(name = "delimiterChar={0}")
+ public static Collection<Object[]> getParams() {
+ return Arrays.asList(new Object[][] {{'/'}, {'.'}});
+ }
+
+ public ResourceNamesTest(char delimiterChar) {
+ super();
+ this.delimiterChar = delimiterChar;
+ delimiter = "" + delimiterChar;
+ testAddress = SimpleString.toSimpleString(UUID.randomUUID().toString());
+ prefix = ActiveMQDefaultConfiguration.getInternalNamingPrefix().replace('.', delimiterChar);
+ baseName = prefix + testAddress + delimiter;
+ testResourceAddressName = baseName + ResourceNames.ADDRESS.replace('.', delimiterChar) + ResourceNames.RETROACTIVE_SUFFIX;
+ testResourceMulticastQueueName = baseName + ResourceNames.QUEUE.replace('.', delimiterChar) + RoutingType.MULTICAST.toString().toLowerCase() + delimiter + ResourceNames.RETROACTIVE_SUFFIX;
+ testResourceAnycastQueueName = baseName + ResourceNames.QUEUE.replace('.', delimiterChar) + RoutingType.ANYCAST.toString().toLowerCase() + delimiter + ResourceNames.RETROACTIVE_SUFFIX;
+ testResourceDivertName = baseName + ResourceNames.DIVERT.replace('.', delimiterChar) + ResourceNames.RETROACTIVE_SUFFIX;
+ }
+
+ @Test
+ public void testGetRetroactiveResourceAddressName() {
+ assertEquals(testResourceAddressName, ResourceNames.getRetroactiveResourceAddressName(prefix, delimiter, testAddress).toString());
+ }
+
+ @Test
+ public void testGetRetroactiveResourceQueueName() {
+ assertEquals(testResourceMulticastQueueName, ResourceNames.getRetroactiveResourceQueueName(prefix, delimiter, testAddress, RoutingType.MULTICAST).toString());
+ assertEquals(testResourceAnycastQueueName, ResourceNames.getRetroactiveResourceQueueName(prefix, delimiter, testAddress, RoutingType.ANYCAST).toString());
+ }
+
+ @Test
+ public void testGetRetroactiveResourceDivertName() {
+ assertEquals(testResourceDivertName, ResourceNames.getRetroactiveResourceDivertName(prefix, delimiter, testAddress).toString());
+ }
+
+ @Test
+ public void testDecomposeRetroactiveResourceAddressName() {
+ assertEquals(testAddress.toString(), ResourceNames.decomposeRetroactiveResourceAddressName(prefix, delimiter, testResourceAddressName));
+ }
+
+ @Test
+ public void testIsRetroactiveResource() {
+ assertTrue(ResourceNames.isRetroactiveResource(prefix, SimpleString.toSimpleString(testResourceAddressName)));
+ assertTrue(ResourceNames.isRetroactiveResource(prefix, SimpleString.toSimpleString(testResourceMulticastQueueName)));
+ assertTrue(ResourceNames.isRetroactiveResource(prefix, SimpleString.toSimpleString(testResourceDivertName)));
+ }
+}
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 a2e42d2..9983ef0 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
@@ -270,6 +270,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
private static final String DEFAULT_RING_SIZE = "default-ring-size";
+ private static final String RETROACTIVE_MESSAGE_COUNT = "retroactive-message-count";
+
// Attributes ----------------------------------------------------
@@ -1174,6 +1176,10 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
addressSettings.setDefaultConsumerWindowSize(XMLUtil.parseInt(child));
} else if (DEFAULT_RING_SIZE.equalsIgnoreCase(name)) {
addressSettings.setDefaultRingSize(XMLUtil.parseLong(child));
+ } else if (RETROACTIVE_MESSAGE_COUNT.equalsIgnoreCase(name)) {
+ long retroactiveMessageCount = XMLUtil.parseLong(child);
+ Validators.GE_ZERO.validate(RETROACTIVE_MESSAGE_COUNT, retroactiveMessageCount);
+ addressSettings.setRetroactiveMessageCount(retroactiveMessageCount);
}
}
return setting;
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 ae94afc..87641cc 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
@@ -2752,6 +2752,7 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
.add("autoDeleteQueuesMessageCount", addressSettings.getAutoDeleteQueuesMessageCount())
.add("autoDeleteAddressesDelay", addressSettings.getAutoDeleteAddressesDelay())
.add("redeliveryCollisionAvoidanceFactor", addressSettings.getRedeliveryCollisionAvoidanceFactor())
+ .add("retroactiveMessageCount", addressSettings.getRetroactiveMessageCount())
.build()
.toString();
}
@@ -2863,12 +2864,12 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
AddressSettings.DEFAULT_CONFIG_DELETE_QUEUES.toString(),
AddressSettings.DEFAULT_CONFIG_DELETE_ADDRESSES.toString(),
AddressSettings.DEFAULT_ADDRESS_REJECT_THRESHOLD,
- ActiveMQDefaultConfiguration.getDefaultLastValueKey().toString(),
+ ActiveMQDefaultConfiguration.getDefaultLastValueKey() == null ? null : ActiveMQDefaultConfiguration.getDefaultLastValueKey().toString(),
ActiveMQDefaultConfiguration.getDefaultNonDestructive(),
ActiveMQDefaultConfiguration.getDefaultExclusive(),
ActiveMQDefaultConfiguration.getDefaultGroupRebalance(),
ActiveMQDefaultConfiguration.getDefaultGroupBuckets(),
- ActiveMQDefaultConfiguration.getDefaultGroupFirstKey().toString(),
+ ActiveMQDefaultConfiguration.getDefaultGroupFirstKey() == null ? null : ActiveMQDefaultConfiguration.getDefaultGroupFirstKey().toString(),
ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(),
ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(),
ActiveMQDefaultConfiguration.getDefaultConsumersBeforeDispatch(),
@@ -2881,7 +2882,8 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
AddressSettings.DEFAULT_AUTO_DELETE_QUEUES_DELAY,
AddressSettings.DEFAULT_AUTO_DELETE_QUEUES_MESSAGE_COUNT,
AddressSettings.DEFAULT_AUTO_DELETE_ADDRESSES_DELAY,
- AddressSettings.DEFAULT_REDELIVER_COLLISION_AVOIDANCE_FACTOR);
+ AddressSettings.DEFAULT_REDELIVER_COLLISION_AVOIDANCE_FACTOR,
+ ActiveMQDefaultConfiguration.getDefaultRetroactiveMessageCount());
}
@Override
@@ -2932,7 +2934,8 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
final long autoDeleteQueuesDelay,
final long autoDeleteQueuesMessageCount,
final long autoDeleteAddressesDelay,
- final double redeliveryCollisionAvoidanceFactor) throws Exception {
+ final double redeliveryCollisionAvoidanceFactor,
+ final long retroactiveMessageCount) throws Exception {
if (AuditLogger.isEnabled()) {
AuditLogger.addAddressSettings(this.server, address, DLA, expiryAddress, expiryDelay, defaultLastValueQueue, maxDeliveryAttempts,
maxSizeBytes, pageSizeBytes, pageMaxCacheSize, redeliveryDelay, redeliveryMultiplier,
@@ -2944,7 +2947,7 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
defaultGroupFirstKey, defaultMaxConsumers, defaultPurgeOnNoConsumers, defaultConsumersBeforeDispatch,
defaultDelayBeforeDispatch, defaultQueueRoutingType, defaultAddressRoutingType, defaultConsumerWindowSize,
defaultRingSize, autoDeleteCreatedQueues, autoDeleteQueuesDelay, autoDeleteQueuesMessageCount,
- autoDeleteAddressesDelay, redeliveryCollisionAvoidanceFactor);
+ autoDeleteAddressesDelay, redeliveryCollisionAvoidanceFactor, retroactiveMessageCount);
}
checkStarted();
@@ -3005,6 +3008,7 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
addressSettings.setAutoDeleteQueuesMessageCount(autoDeleteQueuesMessageCount);
addressSettings.setAutoDeleteAddressesDelay(autoDeleteAddressesDelay);
addressSettings.setRedeliveryCollisionAvoidanceFactor(redeliveryCollisionAvoidanceFactor);
+ addressSettings.setRetroactiveMessageCount(retroactiveMessageCount);
server.getAddressSettingsRepository().addMatch(address, addressSettings);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/AddressControlImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/AddressControlImpl.java
index d2ad001..5332c01 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/AddressControlImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/AddressControlImpl.java
@@ -419,6 +419,14 @@ public class AddressControlImpl extends AbstractControl implements AddressContro
return addressInfo.isPaused();
}
+ @Override
+ public boolean isRetroactiveResource() {
+ if (AuditLogger.isEnabled()) {
+ AuditLogger.isRetroactiveResource(this.addressInfo);
+ }
+ return ResourceNames.isRetroactiveResource(server.getInternalNamingPrefix(), addressInfo.getName());
+ }
+
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/DivertControlImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/DivertControlImpl.java
index dd07b42..6252439 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/DivertControlImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/DivertControlImpl.java
@@ -23,6 +23,7 @@ import java.util.Map;
import org.apache.activemq.artemis.api.core.JsonUtil;
import org.apache.activemq.artemis.api.core.management.DivertControl;
+import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.server.Divert;
@@ -38,6 +39,8 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
private final DivertConfiguration configuration;
+ private final String internalNamingPrefix;
+
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
@@ -46,10 +49,12 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
public DivertControlImpl(final Divert divert,
final StorageManager storageManager,
- final DivertConfiguration configuration) throws Exception {
+ final DivertConfiguration configuration,
+ final String internalNamingPrefix) throws Exception {
super(DivertControl.class, storageManager);
this.divert = divert;
this.configuration = configuration;
+ this.internalNamingPrefix = internalNamingPrefix;
}
@Override
@@ -178,6 +183,14 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
}
@Override
+ public boolean isRetroactiveResource() {
+ if (AuditLogger.isEnabled()) {
+ AuditLogger.isRetroactiveResource(this.divert);
+ }
+ return ResourceNames.isRetroactiveResource(internalNamingPrefix, divert.getUniqueName());
+ }
+
+ @Override
protected MBeanOperationInfo[] fillMBeanOperationInfo() {
return MBeanInfoHelper.getMBeanOperationsInfo(DivertControl.class);
}
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 3f79138..d5a3796 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
@@ -35,6 +35,7 @@ import org.apache.activemq.artemis.api.core.JsonUtil;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.management.QueueControl;
+import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.core.filter.Filter;
import org.apache.activemq.artemis.core.filter.impl.FilterImpl;
import org.apache.activemq.artemis.core.management.impl.openmbean.OpenTypeSupport;
@@ -235,6 +236,21 @@ public class QueueControlImpl extends AbstractControl implements QueueControl {
}
@Override
+ public boolean isRetroactiveResource() {
+ if (AuditLogger.isEnabled()) {
+ AuditLogger.isRetroactiveResource(queue);
+ }
+ checkStarted();
+
+ clearIO();
+ try {
+ return ResourceNames.isRetroactiveResource(server.getInternalNamingPrefix(), queue.getName());
+ } finally {
+ blockOnIO();
+ }
+ }
+
+ @Override
public long getMessageCount() {
if (AuditLogger.isEnabled()) {
AuditLogger.getMessageCount(queue);
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 9465872..029356f 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
@@ -42,6 +42,8 @@ import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.management.CoreNotificationType;
import org.apache.activemq.artemis.api.core.management.ManagementHelper;
import org.apache.activemq.artemis.api.core.management.NotificationType;
+import org.apache.activemq.artemis.api.core.management.ResourceNames;
+import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.config.WildcardConfiguration;
import org.apache.activemq.artemis.core.filter.Filter;
import org.apache.activemq.artemis.core.io.IOCallback;
@@ -63,6 +65,7 @@ import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle;
import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
+import org.apache.activemq.artemis.core.server.ComponentConfigurationRoutingType;
import org.apache.activemq.artemis.core.server.LargeServerMessage;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.Queue;
@@ -78,6 +81,7 @@ import org.apache.activemq.artemis.core.server.management.ManagementService;
import org.apache.activemq.artemis.core.server.management.Notification;
import org.apache.activemq.artemis.core.server.management.NotificationListener;
import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
+import org.apache.activemq.artemis.core.settings.HierarchicalRepositoryChangeListener;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.core.transaction.TransactionOperation;
@@ -459,6 +463,13 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
if (server.hasBrokerAddressPlugins()) {
server.callBrokerAddressPlugins(plugin -> plugin.afterAddAddress(addressInfo, reload));
}
+ long retroactiveMessageCount = addressSettingsRepository.getMatch(addressInfo.getName().toString()).getRetroactiveMessageCount();
+ if (retroactiveMessageCount > 0) {
+ createRetroactiveResources(addressInfo.getName(), retroactiveMessageCount, reload);
+ }
+ if (ResourceNames.isRetroactiveResource(server.getInternalNamingPrefix(), addressInfo.getName())) {
+ registerRepositoryListenerForRetroactiveAddress(addressInfo.getName());
+ }
} catch (Exception e) {
e.printStackTrace();
}
@@ -467,6 +478,127 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
}
}
+ private void registerRepositoryListenerForRetroactiveAddress(SimpleString name) {
+ HierarchicalRepositoryChangeListener repositoryChangeListener = () -> {
+ String prefix = server.getInternalNamingPrefix();
+ String delimiter = server.getConfiguration().getWildcardConfiguration().getDelimiterString();
+ String address = ResourceNames.decomposeRetroactiveResourceAddressName(prefix, delimiter, name.toString());
+ AddressSettings settings = addressSettingsRepository.getMatch(address);
+ Queue internalAnycastQueue = server.locateQueue(ResourceNames.getRetroactiveResourceQueueName(prefix, delimiter, SimpleString.toSimpleString(address), RoutingType.ANYCAST));
+ if (internalAnycastQueue != null && internalAnycastQueue.getRingSize() != settings.getRetroactiveMessageCount()) {
+ internalAnycastQueue.setRingSize(settings.getRetroactiveMessageCount());
+ }
+ Queue internalMulticastQueue = server.locateQueue(ResourceNames.getRetroactiveResourceQueueName(prefix, delimiter, SimpleString.toSimpleString(address), RoutingType.MULTICAST));
+ if (internalMulticastQueue != null && internalMulticastQueue.getRingSize() != settings.getRetroactiveMessageCount()) {
+ internalMulticastQueue.setRingSize(settings.getRetroactiveMessageCount());
+ }
+ };
+ addressSettingsRepository.registerListener(repositoryChangeListener);
+ server.getAddressInfo(name).setRepositoryChangeListener(repositoryChangeListener);
+ }
+
+ private void createRetroactiveResources(final SimpleString retroactiveAddressName, final long retroactiveMessageCount, final boolean reload) throws Exception {
+ String prefix = server.getInternalNamingPrefix();
+ String delimiter = server.getConfiguration().getWildcardConfiguration().getDelimiterString();
+ final SimpleString internalAddressName = ResourceNames.getRetroactiveResourceAddressName(prefix, delimiter, retroactiveAddressName);
+ final SimpleString internalAnycastQueueName = ResourceNames.getRetroactiveResourceQueueName(prefix, delimiter, retroactiveAddressName, RoutingType.ANYCAST);
+ final SimpleString internalMulticastQueueName = ResourceNames.getRetroactiveResourceQueueName(prefix, delimiter, retroactiveAddressName, RoutingType.MULTICAST);
+ final SimpleString internalDivertName = ResourceNames.getRetroactiveResourceDivertName(prefix, delimiter, retroactiveAddressName);
+
+ if (!reload) {
+ AddressInfo addressInfo = new AddressInfo(internalAddressName)
+ .addRoutingType(RoutingType.MULTICAST)
+ .addRoutingType(RoutingType.ANYCAST)
+ .setInternal(false);
+ addAddressInfo(addressInfo);
+ AddressSettings addressSettings = addressSettingsRepository.getMatch(internalAddressName.toString());
+ server.createQueue(internalAddressName,
+ RoutingType.MULTICAST,
+ internalMulticastQueueName,
+ null,
+ null,
+ true,
+ false,
+ false,
+ false,
+ false,
+ 0,
+ false,
+ false,
+ false,
+ addressSettings.getDefaultGroupBuckets(),
+ null,
+ addressSettings.isDefaultLastValueQueue(),
+ addressSettings.getDefaultLastValueKey(),
+ false,
+ 0,
+ 0L,
+ false,
+ 0L,
+ 0L,
+ false,
+ retroactiveMessageCount);
+
+ server.createQueue(internalAddressName,
+ RoutingType.ANYCAST,
+ internalAnycastQueueName,
+ null,
+ null,
+ true,
+ false,
+ false,
+ false,
+ false,
+ 0,
+ false,
+ false,
+ false,
+ addressSettings.getDefaultGroupBuckets(),
+ null,
+ addressSettings.isDefaultLastValueQueue(),
+ addressSettings.getDefaultLastValueKey(),
+ false,
+ 0,
+ 0L,
+ false,
+ 0L,
+ 0L,
+ false,
+ retroactiveMessageCount);
+ }
+ server.deployDivert(new DivertConfiguration()
+ .setName(internalDivertName.toString())
+ .setAddress(retroactiveAddressName.toString())
+ .setExclusive(false)
+ .setForwardingAddress(internalAddressName.toString())
+ .setRoutingType(ComponentConfigurationRoutingType.PASS));
+ }
+
+ private void removeRetroactiveResources(SimpleString address) throws Exception {
+ String prefix = server.getInternalNamingPrefix();
+ String delimiter = server.getConfiguration().getWildcardConfiguration().getDelimiterString();
+
+ SimpleString internalDivertName = ResourceNames.getRetroactiveResourceDivertName(prefix, delimiter, address);
+ if (getBinding(internalDivertName) != null) {
+ server.destroyDivert(internalDivertName);
+ }
+
+ SimpleString internalAnycastQueueName = ResourceNames.getRetroactiveResourceQueueName(prefix, delimiter, address, RoutingType.ANYCAST);
+ if (server.locateQueue(internalAnycastQueueName) != null) {
+ server.destroyQueue(internalAnycastQueueName);
+ }
+
+ SimpleString internalMulticastQueueName = ResourceNames.getRetroactiveResourceQueueName(prefix, delimiter, address, RoutingType.MULTICAST);
+ if (server.locateQueue(internalMulticastQueueName) != null) {
+ server.destroyQueue(internalMulticastQueueName);
+ }
+
+ SimpleString internalAddressName = ResourceNames.getRetroactiveResourceAddressName(prefix, delimiter, address);
+ if (server.getAddressInfo(internalAddressName) != null) {
+ server.removeAddressInfo(internalAddressName, null);
+ }
+ }
+
@Override
public QueueBinding updateQueue(SimpleString name,
RoutingType routingType,
@@ -660,6 +792,7 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
}
managementService.unregisterAddress(address);
final AddressInfo addressInfo = addressManager.removeAddressInfo(address);
+ removeRetroactiveResources(address);
if (server.hasBrokerAddressPlugins()) {
server.callBrokerAddressPlugins(plugin -> plugin.afterRemoveAddress(address, addressInfo));
}
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 4cf2845..24904e2 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
@@ -258,7 +258,7 @@ public class SimpleAddressManager implements AddressManager {
}
@Override
- public boolean reloadAddressInfo(AddressInfo addressInfo) throws Exception {
+ public boolean reloadAddressInfo(AddressInfo addressInfo) {
boolean added = addressInfoMap.putIfAbsent(addressInfo.getName(), addressInfo) == null;
if (added) {
addressInfo.registerMeters(metricsManager);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/RoutingContext.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/RoutingContext.java
index 82091f5..139159b 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/RoutingContext.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/RoutingContext.java
@@ -68,7 +68,7 @@ public interface RoutingContext {
void setAddress(SimpleString address);
- void setRoutingType(RoutingType routingType);
+ RoutingContext setRoutingType(RoutingType routingType);
SimpleString getAddress(Message message);
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 c09792a..3628675 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
@@ -3175,9 +3175,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
recoverStoredConfigs();
- Map<Long, AddressBindingInfo> addressBindingInfosMap = new HashMap<>();
-
- journalLoader.initAddresses(addressBindingInfosMap, addressBindingInfos);
+ journalLoader.initAddresses(addressBindingInfos);
Map<Long, QueueBindingInfo> queueBindingInfosMap = new HashMap<>();
@@ -3285,6 +3283,11 @@ public class ActiveMQServerImpl implements ActiveMQServer {
throw ActiveMQMessageBundle.BUNDLE.addressDoesNotExist(address);
}
+ if (addressInfo.getRepositoryChangeListener() != null) {
+ addressSettingsRepository.unRegisterListener(addressInfo.getRepositoryChangeListener());
+ addressInfo.setRepositoryChangeListener(null);
+ }
+
long txID = storageManager.generateID();
storageManager.deleteAddressBinding(txID, addressInfo.getId());
storageManager.commitBindings(txID);
@@ -3446,6 +3449,8 @@ public class ActiveMQServerImpl implements ActiveMQServer {
managementService.registerQueue(queue, queue.getAddress(), storageManager);
}
+ copyRetroactiveMessages(queue);
+
if (hasBrokerQueuePlugins()) {
callBrokerQueuePlugins(plugin -> plugin.afterCreateQueue(queue));
}
@@ -3455,6 +3460,15 @@ public class ActiveMQServerImpl implements ActiveMQServer {
return queue;
}
+ private void copyRetroactiveMessages(Queue queue) throws Exception {
+ if (addressSettingsRepository.getMatch(queue.getAddress().toString()).getRetroactiveMessageCount() > 0) {
+ Queue retroQueue = locateQueue(ResourceNames.getRetroactiveResourceQueueName(getInternalNamingPrefix(), getConfiguration().getWildcardConfiguration().getDelimiterString(), queue.getAddress(), queue.getRoutingType()));
+ if (retroQueue != null && retroQueue instanceof QueueImpl) {
+ ((QueueImpl) retroQueue).rerouteMessages(queue.getName(), queue.getFilter());
+ }
+ }
+ }
+
@Override
public Queue createQueue(final SimpleString address,
final RoutingType routingType,
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 2f14319..81eefdb 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
@@ -33,6 +33,7 @@ import org.apache.activemq.artemis.core.postoffice.QueueBinding;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.metrics.AddressMetricNames;
import org.apache.activemq.artemis.core.server.metrics.MetricsManager;
+import org.apache.activemq.artemis.core.settings.HierarchicalRepositoryChangeListener;
import org.apache.activemq.artemis.utils.CompositeAddress;
import org.apache.activemq.artemis.utils.PrefixUtil;
@@ -65,6 +66,7 @@ public class AddressInfo {
private PostOffice postOffice;
private StorageManager storageManager;
+ private HierarchicalRepositoryChangeListener repositoryChangeListener;
public AddressInfo(SimpleString name) {
this(name, EnumSet.noneOf(RoutingType.class));
@@ -279,8 +281,9 @@ public class AddressInfo {
return this.internal;
}
- public void setInternal(boolean internal) {
+ public AddressInfo setInternal(boolean internal) {
this.internal = internal;
+ return this;
}
public AddressInfo create(SimpleString name, RoutingType routingType) {
@@ -318,6 +321,15 @@ public class AddressInfo {
return unRoutedMessageCountUpdater.get(this);
}
+ public HierarchicalRepositoryChangeListener getRepositoryChangeListener() {
+ return repositoryChangeListener;
+ }
+
+ public AddressInfo setRepositoryChangeListener(HierarchicalRepositoryChangeListener repositoryChangeListener) {
+ this.repositoryChangeListener = repositoryChangeListener;
+ return this;
+ }
+
public void registerMeters(MetricsManager metricsManager) {
if (metricsManager != null) {
metricsManager.registerAddressGauge(name.toString(), builder -> {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java
index 07997c5..7e37adb 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java
@@ -130,7 +130,7 @@ public class DivertImpl implements Divert {
copy = message;
}
- postOffice.route(copy, new RoutingContextImpl(context.getTransaction()).setReusable(false), false);
+ postOffice.route(copy, new RoutingContextImpl(context.getTransaction()).setReusable(false).setRoutingType(copy.getRoutingType()), false);
}
@Override
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/JournalLoader.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/JournalLoader.java
index 4adb1b2..4c23f8d 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/JournalLoader.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/JournalLoader.java
@@ -38,8 +38,7 @@ public interface JournalLoader {
void initQueues(Map<Long, QueueBindingInfo> queueBindingInfosMap,
List<QueueBindingInfo> queueBindingInfos) throws Exception;
- void initAddresses(Map<Long, AddressBindingInfo> addressBindingInfosMap,
- List<AddressBindingInfo> addressBindingInfo) throws Exception;
+ void initAddresses(List<AddressBindingInfo> addressBindingInfo) throws Exception;
void handleAddMessage(Map<Long, Map<Long, AddMessageRecord>> queueMap) throws Exception;
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 3d73015..60f3528 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
@@ -186,12 +186,9 @@ public class PostOfficeJournalLoader implements JournalLoader {
}
@Override
- public void initAddresses(Map<Long, AddressBindingInfo> addressBindingInfosMap,
- List<AddressBindingInfo> addressBindingInfos) throws Exception {
+ public void initAddresses(List<AddressBindingInfo> addressBindingInfos) throws Exception {
for (AddressBindingInfo addressBindingInfo : addressBindingInfos) {
- addressBindingInfosMap.put(addressBindingInfo.getId(), addressBindingInfo);
-
AddressInfo addressInfo = new AddressInfo(addressBindingInfo.getName()).setRoutingTypes(addressBindingInfo.getRoutingTypes());
addressInfo.setId(addressBindingInfo.getId());
if (addressBindingInfo.getAddressStatusEncoding() != null && addressBindingInfo.getAddressStatusEncoding().getStatus() == AddressQueueStatus.PAUSED) {
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 f21e7ac..d110377 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
@@ -1954,6 +1954,23 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
private int iterQueue(final int flushLimit,
final Filter filter1,
QueueIterateAction messageAction) throws Exception {
+ return iterQueue(flushLimit, filter1, messageAction, true);
+ }
+
+ /**
+ * This is a generic method for any method interacting on the Queue to move or delete messages
+ * Instead of duplicate the feature we created an abstract class where you pass the logic for
+ * each message.
+ *
+ * @param filter1
+ * @param messageAction
+ * @return
+ * @throws Exception
+ */
+ private synchronized int iterQueue(final int flushLimit,
+ final Filter filter1,
+ QueueIterateAction messageAction,
+ final boolean remove) throws Exception {
int count = 0;
int txCount = 0;
@@ -1974,7 +1991,9 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
if (filter1 == null || filter1.match(ref.getMessage())) {
messageAction.actMessage(tx, ref);
- iter.remove();
+ if (remove) {
+ iter.remove();
+ }
txCount++;
count++;
}
@@ -2399,6 +2418,18 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
});
}
+ public synchronized int rerouteMessages(final SimpleString queueName, final Filter filter) throws Exception {
+ return iterQueue(DEFAULT_FLUSH_LIMIT, filter, new QueueIterateAction() {
+ @Override
+ public void actMessage(Transaction tx, MessageReference ref) throws Exception {
+ RoutingContext routingContext = new RoutingContextImpl(tx);
+ routingContext.setAddress(server.locateQueue(queueName).getAddress());
+ server.getPostOffice().getBinding(queueName).route(ref.getMessage(), routingContext);
+ postOffice.processRoute(ref.getMessage(), routingContext, false);
+ }
+ }, false);
+ }
+
@Override
public int retryMessages(Filter filter) throws Exception {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java
index c63f524..563ed4b 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java
@@ -196,11 +196,12 @@ public final class RoutingContextImpl implements RoutingContext {
}
@Override
- public void setRoutingType(RoutingType routingType) {
+ public RoutingContext setRoutingType(RoutingType routingType) {
if (this.routingType == null || this.routingType != routingType) {
this.clear();
}
this.routingType = routingType;
+ return this;
}
@Override
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
index aa1479a..958f9ed 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
@@ -282,7 +282,7 @@ public class ManagementServiceImpl implements ManagementService {
@Override
public synchronized void registerDivert(final Divert divert, final DivertConfiguration config) throws Exception {
ObjectName objectName = objectNameBuilder.getDivertObjectName(divert.getUniqueName().toString(), config.getAddress());
- DivertControl divertControl = new DivertControlImpl(divert, storageManager, config);
+ DivertControl divertControl = new DivertControlImpl(divert, storageManager, config, messagingServer.getInternalNamingPrefix());
registerInJMX(objectName, divertControl);
registerInRegistry(ResourceNames.DIVERT + config.getName(), divertControl);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java
index 6e61683..4ded874 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java
@@ -185,6 +185,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
private Long defaultRingSize = null;
+ private Long retroactiveMessageCount = null;
+
private DeletionPolicy configDeleteQueues = null;
private Boolean autoCreateAddresses = null;
@@ -759,6 +761,15 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
return this;
}
+ public long getRetroactiveMessageCount() {
+ return retroactiveMessageCount != null ? retroactiveMessageCount : ActiveMQDefaultConfiguration.DEFAULT_RETROACTIVE_MESSAGE_COUNT;
+ }
+
+ public AddressSettings setRetroactiveMessageCount(final long defaultRetroactiveMessageCount) {
+ this.retroactiveMessageCount = defaultRetroactiveMessageCount;
+ return this;
+ }
+
/**
* merge 2 objects in to 1
*
@@ -919,6 +930,9 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
if (defaultRingSize == null) {
defaultRingSize = merged.defaultRingSize;
}
+ if (retroactiveMessageCount == null) {
+ retroactiveMessageCount = merged.retroactiveMessageCount;
+ }
}
@Override
@@ -1087,6 +1101,10 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
if (buffer.readableBytes() > 0) {
defaultGroupFirstKey = buffer.readNullableSimpleString();
}
+
+ if (buffer.readableBytes() > 0) {
+ retroactiveMessageCount = BufferHelper.readNullableLong(buffer);
+ }
}
@Override
@@ -1139,7 +1157,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
SimpleString.sizeofNullableString(defaultGroupFirstKey) +
BufferHelper.sizeOfNullableLong(autoDeleteQueuesMessageCount) +
BufferHelper.sizeOfNullableBoolean(autoDeleteCreatedQueues) +
- BufferHelper.sizeOfNullableLong(defaultRingSize);
+ BufferHelper.sizeOfNullableLong(defaultRingSize) +
+ BufferHelper.sizeOfNullableLong(retroactiveMessageCount);
}
@Override
@@ -1243,6 +1262,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
BufferHelper.writeNullableDouble(buffer, redeliveryCollisionAvoidanceFactor);
buffer.writeNullableSimpleString(defaultGroupFirstKey);
+
+ BufferHelper.writeNullableLong(buffer, retroactiveMessageCount);
}
/* (non-Javadoc)
@@ -1303,6 +1324,7 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
result = prime * result + ((defaultGroupBuckets == null) ? 0 : defaultGroupBuckets.hashCode());
result = prime * result + ((defaultGroupFirstKey == null) ? 0 : defaultGroupFirstKey.hashCode());
result = prime * result + ((defaultRingSize == null) ? 0 : defaultRingSize.hashCode());
+ result = prime * result + ((retroactiveMessageCount == null) ? 0 : retroactiveMessageCount.hashCode());
return result;
}
@@ -1585,6 +1607,12 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
return false;
} else if (!defaultRingSize.equals(other.defaultRingSize))
return false;
+
+ if (retroactiveMessageCount == null) {
+ if (other.retroactiveMessageCount != null)
+ return false;
+ } else if (!retroactiveMessageCount.equals(other.retroactiveMessageCount))
+ return false;
return true;
}
@@ -1692,6 +1720,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
defaultGroupFirstKey +
", defaultRingSize=" +
defaultRingSize +
+ ", retroactiveMessageCount=" +
+ retroactiveMessageCount +
"]";
}
}
diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
index 529c39c..f99ca90 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -3398,6 +3398,14 @@
</xsd:documentation>
</xsd:annotation>
</xsd:element>
+
+ <xsd:element name="retroactive-message-count" type="xsd:long" default="0" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the number of messages to preserve for future queues created on the matching address
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
</xsd:all>
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 e0f6372..2bac659 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
@@ -362,6 +362,7 @@ public class FileConfigurationTest extends ConfigurationImplTest {
assertEquals(RoutingType.ANYCAST, conf.getAddressesSettings().get("a1").getDefaultQueueRoutingType());
assertEquals(RoutingType.MULTICAST, conf.getAddressesSettings().get("a1").getDefaultAddressRoutingType());
assertEquals(3, conf.getAddressesSettings().get("a1").getDefaultRingSize());
+ assertEquals(0, conf.getAddressesSettings().get("a1").getRetroactiveMessageCount());
assertEquals("a2.1", conf.getAddressesSettings().get("a2").getDeadLetterAddress().toString());
assertEquals("a2.2", conf.getAddressesSettings().get("a2").getExpiryAddress().toString());
@@ -386,6 +387,7 @@ public class FileConfigurationTest extends ConfigurationImplTest {
assertEquals(RoutingType.ANYCAST, conf.getAddressesSettings().get("a2").getDefaultAddressRoutingType());
assertEquals(10000, conf.getAddressesSettings().get("a2").getDefaultConsumerWindowSize());
assertEquals(-1, conf.getAddressesSettings().get("a2").getDefaultRingSize());
+ assertEquals(10, conf.getAddressesSettings().get("a2").getRetroactiveMessageCount());
assertTrue(conf.getResourceLimitSettings().containsKey("myUser"));
assertEquals(104, conf.getResourceLimitSettings().get("myUser").getMaxConnections());
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/AddressBindingEncodingTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/AddressBindingEncodingTest.java
new file mode 100644
index 0000000..9c89899
--- /dev/null
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/AddressBindingEncodingTest.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright The Apache Software Foundation.
+ *
+ * Licensed 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;
+
+import java.util.EnumSet;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PersistentAddressBindingEncoding;
+import org.apache.activemq.artemis.utils.RandomUtil;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class AddressBindingEncodingTest extends Assert {
+
+ @Test
+ public void testEncodeDecode() {
+ final SimpleString name = RandomUtil.randomSimpleString();
+ final boolean autoCreated = RandomUtil.randomBoolean();
+ final EnumSet<RoutingType> routingTypes = EnumSet.of(RoutingType.ANYCAST, RoutingType.MULTICAST);
+
+ PersistentAddressBindingEncoding encoding = new PersistentAddressBindingEncoding(name,
+ routingTypes,
+ autoCreated);
+ int size = encoding.getEncodeSize();
+ ActiveMQBuffer encodedBuffer = ActiveMQBuffers.fixedBuffer(size);
+ encoding.encode(encodedBuffer);
+
+ PersistentAddressBindingEncoding decoding = new PersistentAddressBindingEncoding();
+ decoding.decode(encodedBuffer);
+
+ assertEquals(name, decoding.getName());
+ assertEquals(autoCreated, decoding.autoCreated);
+ assertEquals(routingTypes, decoding.routingTypes);
+ }
+}
diff --git a/artemis-server/src/test/resources/ConfigurationTest-full-config.xml b/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
index d0843cd..d282d1d 100644
--- a/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
+++ b/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
@@ -431,6 +431,7 @@
<default-queue-routing-type>MULTICAST</default-queue-routing-type>
<default-address-routing-type>ANYCAST</default-address-routing-type>
<default-consumer-window-size>10000</default-consumer-window-size>
+ <retroactive-message-count>10</retroactive-message-count>
</address-setting>
</address-settings>
<resource-limit-settings>
diff --git a/artemis-server/src/test/resources/ConfigurationTest-xinclude-config-address-settings.xml b/artemis-server/src/test/resources/ConfigurationTest-xinclude-config-address-settings.xml
index 92dee3f..8b5e9a2 100644
--- a/artemis-server/src/test/resources/ConfigurationTest-xinclude-config-address-settings.xml
+++ b/artemis-server/src/test/resources/ConfigurationTest-xinclude-config-address-settings.xml
@@ -67,5 +67,6 @@
<default-queue-routing-type>MULTICAST</default-queue-routing-type>
<default-address-routing-type>ANYCAST</default-address-routing-type>
<default-consumer-window-size>10000</default-consumer-window-size>
+ <retroactive-message-count>10</retroactive-message-count>
</address-setting>
</address-settings>
\ No newline at end of file
diff --git a/artemis-tools/src/test/resources/artemis-configuration.xsd b/artemis-tools/src/test/resources/artemis-configuration.xsd
index 6b3f261..891df2d 100644
--- a/artemis-tools/src/test/resources/artemis-configuration.xsd
+++ b/artemis-tools/src/test/resources/artemis-configuration.xsd
@@ -3364,6 +3364,14 @@
</xsd:annotation>
</xsd:element>
+ <xsd:element name="retroactive-message-count" type="xsd:long" default="0" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the number of messages to preserve for future queues created on the matching address
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
</xsd:all>
<xsd:attribute name="match" type="xsd:string" use="required">
@@ -3537,6 +3545,13 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
+ <xsd:attribute name="retroactive-message-count" type="xsd:string" use="optional" default="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ The number of messages to preserve for future queues created on this address
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:attribute>
<xsd:attributeGroup ref="xml:specialAttrs"/>
</xsd:complexType>
diff --git a/docs/user-manual/en/SUMMARY.md b/docs/user-manual/en/SUMMARY.md
index 6cc0efd..5f14233 100644
--- a/docs/user-manual/en/SUMMARY.md
+++ b/docs/user-manual/en/SUMMARY.md
@@ -40,6 +40,7 @@
* [Scheduled Messages](scheduled-messages.md)
* [Last-Value Queues](last-value-queues.md)
* [Ring Queues](ring-queues.md)
+* [Retroactive Addresses](retroactive-addresses.md)
* [Exclusive Queues](exclusive-queues.md)
* [Message Grouping](message-grouping.md)
* [Consumer Priority](consumer-priority.md)
diff --git a/docs/user-manual/en/address-model.md b/docs/user-manual/en/address-model.md
index c19da8a..0a6b50a 100644
--- a/docs/user-manual/en/address-model.md
+++ b/docs/user-manual/en/address-model.md
@@ -613,6 +613,7 @@ that would be found in the `broker.xml` file.
<default-queue-routing-type></default-queue-routing-type>
<default-address-routing-type></default-address-routing-type>
<default-ring-size>-1</default-ring-size>
+ <retroactive-message-count>0</retroactive-message-count>
</address-setting>
</address-settings>
```
@@ -855,8 +856,13 @@ client and/or protocol semantics. Default is `MULTICAST`. Read more about
`default-consumer-window-size` defines the default `consumerWindowSize` value
for a `CORE` protocol consumer, if not defined the default will be set to
1 MiB (1024 * 1024 bytes). The consumer will use this value as the window size
-if the value is not set on the client. Read more about [flow control](#flow-control).
+if the value is not set on the client. Read more about
+[flow control](flow-control.md).
`default-ring-size` defines the default `ring-size` value for any matching queue
which doesn't have `ring-size` explicitly defined. If not defined the default will
-be set to -1. Read more about [ring queues](#ring-queue).
+be set to -1. Read more about [ring queues](ring-queues.md).
+
+`retroactive-message-count` defines the number of messages to preserve for future
+queues created on the matching address. Defaults to 0. Read more about
+[retroactive addresses](retroactive-addresses.md).
diff --git a/docs/user-manual/en/configuration-index.md b/docs/user-manual/en/configuration-index.md
index d244b76..585677c 100644
--- a/docs/user-manual/en/configuration-index.md
+++ b/docs/user-manual/en/configuration-index.md
@@ -248,6 +248,7 @@ Name | Description | Default
[default-queue-routing-type](address-model.md#routing-type) | Routing type for auto-created queues if the type can't be otherwise determined | `MULTICAST`
[default-address-routing-type](address-model.md#routing-type) | Routing type for auto-created addresses if the type can't be otherwise determined | `MULTICAST`
[default-ring-size](ring-queues.md) | The ring-size applied to queues without an explicit `ring-size` configured | `-1`
+[retroactive-message-count](retroactive-addresses.md) | the number of messages to preserve for future queues created on the matching address | `0`
## bridge type
diff --git a/docs/user-manual/en/retroactive-addresses.md b/docs/user-manual/en/retroactive-addresses.md
new file mode 100644
index 0000000..5a7e9da
--- /dev/null
+++ b/docs/user-manual/en/retroactive-addresses.md
@@ -0,0 +1,88 @@
+# Retroactive Addresses
+
+A "retroactive" address is an address that will preserve messages sent to it
+for queues which will be created on it in the future. This can be useful in,
+for example, publish-subscribe use cases where clients want to receive the
+messages sent to the address *before* they they actually connected and created
+their multicast "subscription" queue. Typically messages sent to an address
+before a queue was created on it would simply be unavailable to those queues,
+but with a retroactive address a fixed number of messages can be preserved by
+the broker and automatically copied into queues subsequently created on the
+address. This works for both anycast and multicast queues.
+
+## Internal Retroactive Resources
+
+To implement this functionality the broker will create 4 internal resources for
+each retroactive address:
+
+1. A non-exclusive [divert](#diverts) to grab the messages from the retroactive
+ address.
+2. An address to receive the messages from the divert.
+3. **Two** [ring queues](#ring-queues) to hold the messages sent to the address
+ by the divert - one for anycast and one for multicast. The general caveats
+ for ring queues still apply here. See [the chapter on ring queues](#ring-queues)
+ for more details.
+
+These resources are important to be aware of as they will show up in the web
+console and other management or metric views. They will be named according to
+the following pattern:
+
+```
+<internal-naming-prefix><delimiter><source-address><delimiter>(divert|address|queue<delimiter>(anycast|multicast))<delimiter>retro
+```
+
+For example, if an address named `myAddress` had a `retroactive-message-count`
+of 10 and the default `internal-naming-prefix` (i.e. `$.artemis.internal.`) and
+the default delimiter (i.e. `.`) were being used then resources with these names
+would be created:
+
+1. A divert on `myAddress` named `$.artemis.internal.myAddress.divert.retro`
+2. An address named `$.artemis.internal.myAddress.address.retro`
+3. A multicast queue on the address from step #2 named
+ `$.artemis.internal.myAddress.queue.multicast.retro` with a `ring-size` of 10.
+4. An anycast queue on the address from step #2 named
+ `$.artemis.internal.myAddress.queue.anycast.retro` with a `ring-size` of 10.
+
+This pattern is important to note as it allows one to configure address-settings
+if necessary. To configure custom address-settings you'd use a match like:
+
+```
+*.*.*.<source-address>.*.retro
+```
+
+Using the same example as above the `match` would be:
+
+```
+*.*.*.myAddress.*.retro
+```
+
+> Note:
+>
+> Changing the broker's `internal-naming-prefix` once these retroactive
+> resources are created will break the retroactive functionality.
+>
+
+## Configuration
+
+To configure an address to be "retroactive" simply configure the
+`retroactive-message-count` `address-setting` to reflect the number of messages
+you want the broker to preserve, e.g.:
+
+
+```xml
+<address-settings>
+ <address-setting match="orders">
+ <retroactive-message-count>100</retroactive-message-count>
+ </address-setting>
+</address-settings>
+```
+
+The value for `retroactive-message-count` can be updated at runtime either via
+`broker.xml` or via the management API just like any other address-setting.
+However, if you *reduce* the value of `retroactive-message-count` an additional
+administrative step will be required since this functionality is implemented
+via ring queues. This is because a ring queue whose ring-size is reduced will
+not automatically delete messages from the queue to meet the new ring-size in
+order to avoid unintended message loss. Therefore, administrative action will
+be required in this case to manually reduce the number of messages in the ring
+queue via the management API.
\ No newline at end of file
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/RequestReplyMultiProtocolTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/RequestReplyMultiProtocolTest.java
index b842942..d3efc68 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/RequestReplyMultiProtocolTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/RequestReplyMultiProtocolTest.java
@@ -28,14 +28,12 @@ import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.TextMessage;
import javax.jms.Topic;
-
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
-import org.apache.activemq.artemis.core.postoffice.impl.PostOfficeImpl;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.tests.integration.openwire.OpenWireTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
@@ -44,6 +42,7 @@ import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+
import static org.apache.activemq.artemis.tests.util.CFUtil.createConnectionFactory;
@RunWith(Parameterized.class)
@@ -92,7 +91,7 @@ public class RequestReplyMultiProtocolTest extends OpenWireTestBase {
this.server.createQueue(queueName, RoutingType.ANYCAST, queueName, null, true, false, -1, false, true);
this.server.createQueue(replyQueue, RoutingType.ANYCAST, replyQueue, null, true, false, -1, false, true);
AddressInfo info = new AddressInfo(topicName, RoutingType.MULTICAST);
- ((PostOfficeImpl)this.server.getPostOffice()).getAddressManager().addAddressInfo(info);
+ this.server.addAddressInfo(info);
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/RequestReplyNonJMSTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/RequestReplyNonJMSTest.java
index d15ecd8..a523bd2 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/RequestReplyNonJMSTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/RequestReplyNonJMSTest.java
@@ -16,12 +16,6 @@
*/
package org.apache.activemq.artemis.tests.integration.crossprotocol;
-import static org.apache.activemq.artemis.tests.util.CFUtil.createConnectionFactory;
-
-import java.net.URI;
-import java.util.Arrays;
-import java.util.UUID;
-
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
@@ -30,10 +24,12 @@ import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.UUID;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
-import org.apache.activemq.artemis.core.postoffice.impl.PostOfficeImpl;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.tests.integration.openwire.OpenWireTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
@@ -48,6 +44,8 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import static org.apache.activemq.artemis.tests.util.CFUtil.createConnectionFactory;
+
@RunWith(Parameterized.class)
public class RequestReplyNonJMSTest extends OpenWireTestBase {
@@ -84,7 +82,7 @@ public class RequestReplyNonJMSTest extends OpenWireTestBase {
this.server.createQueue(queueName, RoutingType.ANYCAST, queueName, null, true, false, -1, false, true);
this.server.createQueue(replyQueue, RoutingType.ANYCAST, replyQueue, null, true, false, -1, false, true);
AddressInfo info = new AddressInfo(topicName, RoutingType.MULTICAST);
- ((PostOfficeImpl)this.server.getPostOffice()).getAddressManager().addAddressInfo(info);
+ this.server.addAddressInfo(info);
}
@Test
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 8e52cbc..53756ef 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
@@ -727,7 +727,6 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
boolean autoDeleteJmsQueues = RandomUtil.randomBoolean();
boolean autoCreateJmsTopics = RandomUtil.randomBoolean();
boolean autoDeleteJmsTopics = RandomUtil.randomBoolean();
-
boolean autoCreateQueues = RandomUtil.randomBoolean();
boolean autoDeleteQueues = RandomUtil.randomBoolean();
boolean autoCreateAddresses = RandomUtil.randomBoolean();
@@ -754,6 +753,7 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
long autoDeleteQueuesMessageCount = RandomUtil.randomPositiveLong();
long autoDeleteAddressesDelay = RandomUtil.randomPositiveLong();
double redeliveryCollisionAvoidanceFactor = RandomUtil.randomDouble();
+ long retroactiveMessageCount = RandomUtil.randomPositiveLong();
serverControl.addAddressSettings(addressMatch,
DLA,
@@ -802,7 +802,8 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
autoDeleteQueuesDelay,
autoDeleteQueuesMessageCount,
autoDeleteAddressesDelay,
- redeliveryCollisionAvoidanceFactor);
+ redeliveryCollisionAvoidanceFactor,
+ retroactiveMessageCount);
boolean ex = false;
try {
@@ -853,7 +854,8 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
autoDeleteQueuesDelay,
autoDeleteQueuesMessageCount,
autoDeleteAddressesDelay,
- redeliveryCollisionAvoidanceFactor);
+ redeliveryCollisionAvoidanceFactor,
+ retroactiveMessageCount);
} catch (Exception expected) {
ex = true;
}
@@ -911,6 +913,7 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
assertEquals(autoDeleteQueuesMessageCount, info.getAutoDeleteQueuesMessageCount());
assertEquals(autoDeleteAddressesDelay, info.getAutoDeleteAddressesDelay());
assertEquals(redeliveryCollisionAvoidanceFactor, info.getRedeliveryCollisionAvoidanceFactor(), 0);
+ assertEquals(retroactiveMessageCount, info.getRetroactiveMessageCount());
serverControl.addAddressSettings(addressMatch,
DLA,
@@ -959,7 +962,8 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
autoDeleteQueuesDelay,
autoDeleteQueuesMessageCount,
autoDeleteAddressesDelay,
- redeliveryCollisionAvoidanceFactor);
+ redeliveryCollisionAvoidanceFactor,
+ retroactiveMessageCount);
jsonString = serverControl.getAddressSettingsAsJSON(exactAddress);
info = AddressSettingsInfo.from(jsonString);
@@ -1010,6 +1014,7 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
assertEquals(autoDeleteQueuesMessageCount, info.getAutoDeleteQueuesMessageCount());
assertEquals(autoDeleteAddressesDelay, info.getAutoDeleteAddressesDelay());
assertEquals(redeliveryCollisionAvoidanceFactor, info.getRedeliveryCollisionAvoidanceFactor(), 0);
+ assertEquals(retroactiveMessageCount, info.getRetroactiveMessageCount());
ex = false;
try {
@@ -1060,7 +1065,8 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
autoDeleteQueuesDelay,
autoDeleteQueuesMessageCount,
autoDeleteAddressesDelay,
- redeliveryCollisionAvoidanceFactor);
+ redeliveryCollisionAvoidanceFactor,
+ retroactiveMessageCount);
} catch (Exception e) {
ex = true;
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java
index c727b07..d340264 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java
@@ -954,7 +954,8 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
@Parameter(desc = "delay for deleting auto-created queues", name = "autoDeleteQueuesDelay") long autoDeleteQueuesDelay,
@Parameter(desc = "the message count the queue must be at or below before it can be auto deleted", name = "autoDeleteQueuesMessageCount") long autoDeleteQueuesMessageCount,
@Parameter(desc = "delay for deleting auto-created addresses", name = "autoDeleteAddressesDelay") long autoDeleteAddressesDelay,
- @Parameter(desc = "factor by which to modify the redelivery delay slightly to avoid collisions", name = "redeliveryCollisionAvoidanceFactor") double redeliveryCollisionAvoidanceFactor) throws Exception {
+ @Parameter(desc = "factor by which to modify the redelivery delay slightly to avoid collisions", name = "redeliveryCollisionAvoidanceFactor") double redeliveryCollisionAvoidanceFactor,
+ @Parameter(desc = "the number of messages to preserve for future queues created on the matching address", name = "retroactiveMessageCount") long retroactiveMessageCount) throws Exception {
proxy.invokeOperation("addAddressSettings",
addressMatch,
DLA,
@@ -1003,7 +1004,8 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
autoDeleteQueuesDelay,
autoDeleteQueuesMessageCount,
autoDeleteAddressesDelay,
- redeliveryCollisionAvoidanceFactor);
+ redeliveryCollisionAvoidanceFactor,
+ retroactiveMessageCount);
}
@Override
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/AddressControlTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/AddressControlTest.java
index 565ee4c..ad6b34f 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/AddressControlTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/AddressControlTest.java
@@ -37,6 +37,7 @@ import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
import org.apache.activemq.artemis.api.core.client.ServerLocator;
import org.apache.activemq.artemis.api.core.management.AddressControl;
+import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.api.core.management.RoleInfo;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.security.CheckType;
@@ -85,6 +86,18 @@ public class AddressControlTest extends ManagementTestBase {
}
@Test
+ public void testIsRetroactiveResource() throws Exception {
+ SimpleString baseAddress = RandomUtil.randomSimpleString();
+ SimpleString address = ResourceNames.getRetroactiveResourceAddressName(server.getInternalNamingPrefix(), server.getConfiguration().getWildcardConfiguration().getDelimiterString(), baseAddress);
+
+ session.createAddress(address, RoutingType.MULTICAST, false);
+
+ AddressControl addressControl = createManagementControl(address);
+
+ Assert.assertTrue(addressControl.isRetroactiveResource());
+ }
+
+ @Test
public void testGetQueueNames() throws Exception {
SimpleString address = RandomUtil.randomSimpleString();
SimpleString queue = RandomUtil.randomSimpleString();
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/AddressControlUsingCoreTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/AddressControlUsingCoreTest.java
index 9ed7bef..c76f4d7 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/AddressControlUsingCoreTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/AddressControlUsingCoreTest.java
@@ -134,6 +134,11 @@ public class AddressControlUsingCoreTest extends AddressControlTest {
}
@Override
+ public boolean isRetroactiveResource() {
+ return (boolean) proxy.retrieveAttributeValue("retroactiveResource");
+ }
+
+ @Override
public String sendMessage(Map<String, String> headers,
int type,
String body,
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/DivertControlTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/DivertControlTest.java
index 749f626..3c4b983 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/DivertControlTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/DivertControlTest.java
@@ -16,9 +16,11 @@
*/
package org.apache.activemq.artemis.tests.integration.management;
+import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.api.core.management.DivertControl;
import org.apache.activemq.artemis.api.core.management.ObjectNameBuilder;
+import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.CoreQueueConfiguration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
@@ -68,6 +70,28 @@ public class DivertControlTest extends ManagementTestBase {
Assert.assertEquals(divertConfig.getTransformerConfiguration().getProperties(), divertControl.getTransformerProperties());
}
+ @Test
+ public void testRetroactiveResourceAttribute() throws Exception {
+ String address = RandomUtil.randomString();
+ CoreQueueConfiguration queueConfig = new CoreQueueConfiguration().setAddress(RandomUtil.randomString()).setName(RandomUtil.randomString()).setDurable(false);
+ CoreQueueConfiguration forwardQueueConfig = new CoreQueueConfiguration().setAddress(address).setName(RandomUtil.randomString()).setDurable(false);
+
+ divertConfig = new DivertConfiguration()
+ .setName(ResourceNames.getRetroactiveResourceDivertName(server.getInternalNamingPrefix(), server.getConfiguration().getWildcardConfiguration().getDelimiterString(), SimpleString.toSimpleString(address)).toString())
+ .setRoutingName(RandomUtil.randomString()).setAddress(queueConfig.getAddress())
+ .setForwardingAddress(forwardQueueConfig.getAddress())
+ .setExclusive(RandomUtil.randomBoolean())
+ .setTransformerConfiguration(new TransformerConfiguration(AddHeadersTransformer.class.getName()));
+
+ server.deployDivert(divertConfig);
+
+ checkResource(ObjectNameBuilder.DEFAULT.getDivertObjectName(divertConfig.getName(), divertConfig.getAddress()));
+
+ DivertControl divertControl = createDivertManagementControl(divertConfig.getName(), divertConfig.getAddress());
+
+ Assert.assertTrue(divertControl.isRetroactiveResource());
+ }
+
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/DivertControlUsingCoreTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/DivertControlUsingCoreTest.java
index 2705ab6..e7d286a 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/DivertControlUsingCoreTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/DivertControlUsingCoreTest.java
@@ -88,6 +88,11 @@ public class DivertControlUsingCoreTest extends DivertControlTest {
return (Boolean) proxy.retrieveAttributeValue("exclusive");
}
+ @Override
+ public boolean isRetroactiveResource() {
+ return (Boolean) proxy.retrieveAttributeValue("retroactiveResource");
+ }
+
};
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlTest.java
index 32e52cc..4596558 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlTest.java
@@ -56,6 +56,7 @@ import org.apache.activemq.artemis.api.core.management.DayCounterInfo;
import org.apache.activemq.artemis.api.core.management.MessageCounterInfo;
import org.apache.activemq.artemis.api.core.management.ObjectNameBuilder;
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.Configuration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.messagecounter.impl.MessageCounterManagerImpl;
@@ -67,8 +68,8 @@ import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.QueueImpl;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.transaction.impl.XidImpl;
-import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.tests.integration.jms.server.management.JMSUtil;
+import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.Base64;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.junit.Assert;
@@ -122,6 +123,27 @@ public class QueueControlTest extends ManagementTestBase {
}
@Test
+ public void testRetroactiveResourceAttribute() throws Exception {
+ SimpleString baseAddress = RandomUtil.randomSimpleString();
+ String internalNamingPrefix = server.getInternalNamingPrefix();
+ String delimiter = server.getConfiguration().getWildcardConfiguration().getDelimiterString();
+ SimpleString address = ResourceNames.getRetroactiveResourceAddressName(internalNamingPrefix, delimiter, baseAddress);
+ SimpleString multicastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, baseAddress, RoutingType.MULTICAST);
+ SimpleString anycastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, baseAddress, RoutingType.ANYCAST);
+
+ session.createQueue(address, RoutingType.MULTICAST, multicastQueue, null, durable);
+ session.createQueue(address, RoutingType.MULTICAST, anycastQueue, null, durable);
+
+ QueueControl queueControl = createManagementControl(address, multicastQueue);
+ Assert.assertTrue(queueControl.isRetroactiveResource());
+ queueControl = createManagementControl(address, anycastQueue);
+ Assert.assertTrue(queueControl.isRetroactiveResource());
+
+ session.deleteQueue(multicastQueue);
+ session.deleteQueue(anycastQueue);
+ }
+
+ @Test
public void testGetNullFilter() throws Exception {
SimpleString address = RandomUtil.randomSimpleString();
SimpleString queue = RandomUtil.randomSimpleString();
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 f72fb11..2b4f418 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
@@ -311,6 +311,11 @@ public class QueueControlUsingCoreTest extends QueueControlTest {
}
@Override
+ public boolean isRetroactiveResource() {
+ return (Boolean) proxy.retrieveAttributeValue("retroactiveResource");
+ }
+
+ @Override
public String listMessageCounter() throws Exception {
return (String) proxy.invokeOperation("listMessageCounter");
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/RetroactiveAddressFailoverTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/RetroactiveAddressFailoverTest.java
new file mode 100644
index 0000000..4198ca7
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/RetroactiveAddressFailoverTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.config.ActiveMQDefaultConfiguration;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.api.core.TransportConfiguration;
+import org.apache.activemq.artemis.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.ServerLocator;
+import org.apache.activemq.artemis.api.core.management.ResourceNames;
+import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryInternal;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.integration.IntegrationTestLogger;
+import org.apache.activemq.artemis.tests.integration.cluster.failover.FailoverTestBase;
+import org.apache.activemq.artemis.tests.util.TransportConfigurationUtils;
+import org.junit.Before;
+import org.junit.Test;
+
+public class RetroactiveAddressFailoverTest extends FailoverTestBase {
+
+ private static final IntegrationTestLogger log = IntegrationTestLogger.LOGGER;
+
+ protected ServerLocator locator;
+
+ protected ClientSessionFactoryInternal sf;
+
+ String internalNamingPrefix = ActiveMQDefaultConfiguration.DEFAULT_INTERNAL_NAMING_PREFIX;
+
+ String delimiter = ".";
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ locator = getServerLocator().setBlockOnNonDurableSend(true).setBlockOnDurableSend(true).setReconnectAttempts(300).setRetryInterval(100);
+ sf = createSessionFactoryAndWaitForTopology(locator, 2);
+ }
+
+ @Test
+ public void testFailover() throws Exception {
+ final int MESSAGE_COUNT = 10;
+ final int OFFSET = 5;
+ ActiveMQServer live = liveServer.getServer();
+ ActiveMQServer backup = backupServer.getServer();
+ ClientSession session = addClientSession(sf.createSession(true, true));
+ final SimpleString queueName = SimpleString.toSimpleString("simpleQueue");
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ live.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(MESSAGE_COUNT));
+ backup.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(MESSAGE_COUNT));
+ live.addAddressInfo(new AddressInfo(addressName));
+
+ ClientProducer producer = addClientProducer(session.createProducer(addressName));
+ for (int j = 0; j < OFFSET; j++) {
+ ClientMessage message = session.createMessage(true);
+ message.putIntProperty("xxx", j);
+ producer.send(message);
+ }
+
+ org.apache.activemq.artemis.tests.util.Wait.assertTrue(() -> live.locateQueue(divertQueue).getMessageCount() == OFFSET, 3000, 50);
+
+ crash(session);
+
+ org.apache.activemq.artemis.tests.util.Wait.assertTrue(() -> backup.locateQueue(divertQueue).getMessageCount() == OFFSET, 3000, 50);
+
+ for (int j = OFFSET; j < MESSAGE_COUNT + OFFSET; j++) {
+ ClientMessage message = session.createMessage(true);
+ message.putIntProperty("xxx", j);
+ producer.send(message);
+ }
+
+ org.apache.activemq.artemis.tests.util.Wait.assertTrue(() -> backup.locateQueue(divertQueue).getMessageCount() == MESSAGE_COUNT, 3000, 50);
+
+ session.createQueue(addressName.toString(), RoutingType.ANYCAST, queueName.toString());
+ org.apache.activemq.artemis.tests.util.Wait.assertTrue(() -> backup.locateQueue(queueName) != null);
+ org.apache.activemq.artemis.tests.util.Wait.assertTrue(() -> backup.locateQueue(queueName).getMessageCount() == MESSAGE_COUNT, 500, 50);
+
+ ClientConsumer consumer = session.createConsumer(queueName);
+ for (int j = OFFSET; j < MESSAGE_COUNT + OFFSET; j++) {
+ session.start();
+ ClientMessage message = consumer.receive(1000);
+ assertNotNull(message);
+ message.acknowledge();
+ assertEquals(j, (int) message.getIntProperty("xxx"));
+ }
+ consumer.close();
+ session.deleteQueue(queueName);
+ }
+
+ @Override
+ protected TransportConfiguration getAcceptorTransportConfiguration(final boolean live) {
+ return TransportConfigurationUtils.getInVMAcceptor(live);
+ }
+
+ @Override
+ protected TransportConfiguration getConnectorTransportConfiguration(final boolean live) {
+ return TransportConfigurationUtils.getInVMConnector(live);
+ }
+}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/RetroactiveAddressTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/RetroactiveAddressTest.java
new file mode 100644
index 0000000..76d85c4
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/RetroactiveAddressTest.java
@@ -0,0 +1,476 @@
+/*
+ * 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
+ * <br>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <br>
+ * 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 javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.UUID;
+
+import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+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.api.core.management.ResourceNames;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+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.jms.client.ActiveMQConnectionFactory;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class RetroactiveAddressTest extends ActiveMQTestBase {
+
+ protected ActiveMQServer server;
+
+ protected ClientSession session;
+
+ protected ClientSessionFactory sf;
+
+ protected ServerLocator locator;
+
+ String internalNamingPrefix;
+
+ char delimiterChar;
+
+ String delimiter;
+
+ @Parameterized.Parameters(name = "delimiterChar={0}")
+ public static Collection<Object[]> getParams() {
+ return Arrays.asList(new Object[][] {{'/'}, {'.'}});
+ }
+
+ public RetroactiveAddressTest(char delimiterChar) {
+ super();
+ this.delimiterChar = delimiterChar;
+ }
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ server = createServer(true, createDefaultInVMConfig());
+ server.getConfiguration().setInternalNamingPrefix(ActiveMQDefaultConfiguration.DEFAULT_INTERNAL_NAMING_PREFIX.replace('.', delimiterChar));
+ server.getConfiguration().getWildcardConfiguration().setDelimiter(delimiterChar);
+ server.start();
+ locator = createInVMNonHALocator();
+ sf = createSessionFactory(locator);
+ session = addClientSession(sf.createSession(false, true, true));
+ internalNamingPrefix = server.getConfiguration().getInternalNamingPrefix();
+ delimiter = server.getConfiguration().getWildcardConfiguration().getDelimiterString();
+ }
+
+ @Test
+ public void testRetroactiveResourceCreation() throws Exception {
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertAddress = ResourceNames.getRetroactiveResourceAddressName(internalNamingPrefix, delimiter, addressName);
+ final SimpleString divertMulticastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ final SimpleString divertAnycastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.ANYCAST);
+ final SimpleString divert = ResourceNames.getRetroactiveResourceDivertName(internalNamingPrefix, delimiter, addressName);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(10));
+ server.addAddressInfo(new AddressInfo(addressName));
+ assertNotNull(server.getAddressInfo(divertAddress));
+ assertNotNull(server.locateQueue(divertMulticastQueue));
+ assertEquals(RoutingType.MULTICAST, server.locateQueue(divertMulticastQueue).getRoutingType());
+ assertNotNull(server.locateQueue(divertAnycastQueue));
+ assertEquals(RoutingType.ANYCAST, server.locateQueue(divertAnycastQueue).getRoutingType());
+ assertNotNull(server.getPostOffice().getBinding(divert));
+ }
+
+ @Test
+ public void testRetroactiveResourceRemoval() throws Exception {
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertAddress = ResourceNames.getRetroactiveResourceAddressName(internalNamingPrefix, delimiter, addressName);
+ final SimpleString divertMulticastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ final SimpleString divertAnycastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.ANYCAST);
+ final SimpleString divert = ResourceNames.getRetroactiveResourceDivertName(internalNamingPrefix, delimiter, addressName);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(10));
+
+ server.addAddressInfo(new AddressInfo(addressName));
+ assertNotNull(server.getAddressInfo(divertAddress));
+ assertNotNull(server.locateQueue(divertMulticastQueue));
+ assertNotNull(server.locateQueue(divertAnycastQueue));
+ assertNotNull(server.getPostOffice().getBinding(divert));
+
+ server.removeAddressInfo(addressName, null, true);
+ assertNull(server.getAddressInfo(divertAddress));
+ assertNull(server.locateQueue(divertAnycastQueue));
+ assertNull(server.locateQueue(divertMulticastQueue));
+ assertNull(server.getPostOffice().getBinding(divert));
+ }
+
+ @Test
+ public void testRetroactiveAddress() throws Exception {
+ final int COUNT = 15;
+ final int LOOPS = 25;
+ final SimpleString queueName = SimpleString.toSimpleString("simpleQueue");
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(COUNT));
+ server.addAddressInfo(new AddressInfo(addressName));
+
+ for (int i = 0; i < LOOPS; i++) {
+ ClientProducer producer = session.createProducer(addressName);
+ for (int j = 0; j < COUNT; j++) {
+ ClientMessage message = session.createMessage(false);
+ message.putIntProperty("xxx", (i * COUNT) + j);
+ producer.send(message);
+ }
+ producer.close();
+
+ final int finalI = i;
+ Wait.assertTrue(() -> server.locateQueue(divertQueue).getMessagesReplaced() == (COUNT * finalI), 3000, 50);
+ Wait.assertTrue(() -> server.locateQueue(divertQueue).getMessageCount() == COUNT, 3000, 50);
+
+ session.createQueue(addressName.toString(), RoutingType.ANYCAST, queueName.toString());
+ Wait.assertTrue(() -> server.locateQueue(queueName) != null);
+ Wait.assertTrue(() -> server.locateQueue(queueName).getMessageCount() == COUNT, 500, 50);
+ ClientConsumer consumer = session.createConsumer(queueName);
+ for (int j = 0; j < COUNT; j++) {
+ session.start();
+ ClientMessage message = consumer.receive(1000);
+ assertNotNull(message);
+ message.acknowledge();
+ assertEquals((i * COUNT) + j, (int) message.getIntProperty("xxx"));
+ }
+ consumer.close();
+ session.deleteQueue(queueName);
+ }
+ }
+
+ @Test
+ public void testRestart() throws Exception {
+ final String data = "Simple Text " + UUID.randomUUID().toString();
+ final SimpleString queueName1 = SimpleString.toSimpleString("simpleQueue1");
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertMulticastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(10));
+ server.addAddressInfo(new AddressInfo(addressName));
+
+ ClientProducer producer = session.createProducer(addressName);
+ ClientMessage message = session.createMessage(true);
+ message.getBodyBuffer().writeString(data + "1");
+ producer.send(message);
+ producer.close();
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getMessageCount() == 1, 500, 50);
+
+ server.stop();
+ server.start();
+ assertNotNull(server.locateQueue(divertMulticastQueue));
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getMessageCount() == 1, 500, 50);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(10));
+ locator = createInVMNonHALocator();
+ sf = createSessionFactory(locator);
+ session = addClientSession(sf.createSession(false, true, true));
+
+ producer = session.createProducer(addressName);
+ message = session.createMessage(true);
+ message.getBodyBuffer().writeString(data + "2");
+ producer.send(message);
+ producer.close();
+
+ session.createQueue(addressName.toString(), RoutingType.ANYCAST, queueName1.toString());
+ Wait.assertTrue(() -> server.locateQueue(queueName1) != null);
+ Wait.assertTrue(() -> server.locateQueue(queueName1).getMessageCount() == 2, 500, 50);
+
+ ClientConsumer consumer = session.createConsumer(queueName1);
+ session.start();
+ message = consumer.receive(1000);
+ assertNotNull(message);
+ message.acknowledge();
+ assertEquals(data + "1", message.getBodyBuffer().readString());
+ message = consumer.receive(1000);
+ assertNotNull(message);
+ message.acknowledge();
+ assertEquals(data + "2", message.getBodyBuffer().readString());
+ consumer.close();
+ Wait.assertTrue(() -> server.locateQueue(queueName1).getMessageCount() == 0, 500, 50);
+
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getMessageCount() == 2, 2000, 100);
+ }
+
+ @Test
+ public void testUpdateAfterRestart() throws Exception {
+ final int COUNT = 10;
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertAnycastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.ANYCAST);
+ final SimpleString divertMulticastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(COUNT));
+ server.addAddressInfo(new AddressInfo(addressName));
+ Wait.assertTrue(() -> server.locateQueue(divertAnycastQueue).getRingSize() == COUNT, 1000, 100);
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getRingSize() == COUNT, 1000, 100);
+ server.stop();
+ server.start();
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(COUNT * 2));
+ Wait.assertTrue(() -> server.locateQueue(divertAnycastQueue).getRingSize() == COUNT * 2, 1000, 100);
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getRingSize() == COUNT * 2, 1000, 100);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(COUNT));
+ Wait.assertTrue(() -> server.locateQueue(divertAnycastQueue).getRingSize() == COUNT, 1000, 100);
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getRingSize() == COUNT, 1000, 100);
+ }
+
+ @Test
+ public void testMulticast() throws Exception {
+ final String data = "Simple Text " + UUID.randomUUID().toString();
+ final SimpleString queueName1 = SimpleString.toSimpleString("simpleQueue1");
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(10));
+ server.addAddressInfo(new AddressInfo(addressName));
+
+ ClientProducer producer = session.createProducer(addressName);
+ ClientMessage message = session.createMessage(false);
+ message.getBodyBuffer().writeString(data);
+ message.setRoutingType(RoutingType.MULTICAST);
+ producer.send(message);
+ producer.close();
+ Wait.assertTrue(() -> server.locateQueue(divertQueue).getMessageCount() == 1, 500, 50);
+
+ session.createQueue(addressName.toString(), RoutingType.MULTICAST, queueName1.toString());
+ Wait.assertTrue(() -> server.locateQueue(queueName1) != null);
+ Wait.assertTrue(() -> server.locateQueue(queueName1).getMessageCount() == 1, 500, 50);
+
+ ClientConsumer consumer = session.createConsumer(queueName1);
+ session.start();
+ message = consumer.receive(1000);
+ assertNotNull(message);
+ message.acknowledge();
+ assertEquals(data, message.getBodyBuffer().readString());
+ consumer.close();
+ Wait.assertTrue(() -> server.locateQueue(queueName1).getMessageCount() == 0, 500, 50);
+
+ Wait.assertTrue(() -> server.locateQueue(divertQueue).getMessageCount() == 1, 2000, 100);
+ }
+
+ @Test
+ public void testJMSTopicSubscribers() throws Exception {
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final int COUNT = 10;
+ final SimpleString divertQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(COUNT));
+ server.addAddressInfo(new AddressInfo(addressName));
+
+ ConnectionFactory cf = new ActiveMQConnectionFactory("vm://0");
+ Connection c = cf.createConnection();
+ Session s = c.createSession();
+ Topic t = s.createTopic(addressName.toString());
+
+ MessageProducer producer = s.createProducer(t);
+ for (int i = 0; i < COUNT * 2; i++) {
+ Message m = s.createMessage();
+ m.setIntProperty("test", i);
+ producer.send(m);
+ }
+ producer.close();
+ Wait.assertTrue(() -> server.locateQueue(divertQueue).getMessageCount() == COUNT, 500, 50);
+
+ MessageConsumer consumer = s.createConsumer(t);
+ c.start();
+ for (int i = 0; i < COUNT; i++) {
+ Message m = consumer.receive(500);
+ assertNotNull(m);
+ assertEquals(i + COUNT, m.getIntProperty("test"));
+ }
+ assertNull(consumer.receiveNoWait());
+ }
+
+ @Test
+ public void testUpdateAddressSettings() throws Exception {
+ final int COUNT = 10;
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertAnycastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.ANYCAST);
+ final SimpleString divertMulticastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(COUNT));
+ server.addAddressInfo(new AddressInfo(addressName));
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(COUNT * 2));
+ Wait.assertTrue(() -> server.locateQueue(divertAnycastQueue).getRingSize() == COUNT * 2, 1000, 100);
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getRingSize() == COUNT * 2, 1000, 100);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(COUNT));
+ Wait.assertTrue(() -> server.locateQueue(divertAnycastQueue).getRingSize() == COUNT, 1000, 100);
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getRingSize() == COUNT, 1000, 100);
+ }
+
+ @Test
+ public void testRoutingTypes() throws Exception {
+ final String data = "Simple Text " + UUID.randomUUID().toString();
+ final SimpleString multicastQueue = SimpleString.toSimpleString("multicastQueue");
+ final SimpleString anycastQueue = SimpleString.toSimpleString("anycastQueue");
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertMulticastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ final SimpleString divertAnycastQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.ANYCAST);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(10));
+ server.addAddressInfo(new AddressInfo(addressName));
+
+ ClientProducer producer = session.createProducer(addressName);
+ ClientMessage message = session.createMessage(false);
+ message.getBodyBuffer().writeString(data + RoutingType.MULTICAST.toString());
+ message.setRoutingType(RoutingType.MULTICAST);
+ producer.send(message);
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getMessageCount() == 1, 500, 50);
+ Wait.assertTrue(() -> server.locateQueue(divertAnycastQueue).getMessageCount() == 0, 500, 50);
+
+ message = session.createMessage(false);
+ message.getBodyBuffer().writeString(data + RoutingType.ANYCAST.toString());
+ message.setRoutingType(RoutingType.ANYCAST);
+ producer.send(message);
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getMessageCount() == 1, 500, 50);
+ Wait.assertTrue(() -> server.locateQueue(divertAnycastQueue).getMessageCount() == 1, 500, 50);
+
+ producer.close();
+
+ session.createQueue(addressName.toString(), RoutingType.MULTICAST, multicastQueue.toString());
+ Wait.assertTrue(() -> server.locateQueue(multicastQueue) != null);
+ Wait.assertTrue(() -> server.locateQueue(multicastQueue).getMessageCount() == 1, 500, 50);
+
+ session.createQueue(addressName.toString(), RoutingType.ANYCAST, anycastQueue.toString());
+ Wait.assertTrue(() -> server.locateQueue(anycastQueue) != null);
+ Wait.assertTrue(() -> server.locateQueue(anycastQueue).getMessageCount() == 1, 500, 50);
+
+ ClientConsumer consumer = session.createConsumer(multicastQueue);
+ session.start();
+ message = consumer.receive(1000);
+ assertNotNull(message);
+ message.acknowledge();
+ assertEquals(data + RoutingType.MULTICAST.toString(), message.getBodyBuffer().readString());
+ consumer.close();
+ Wait.assertTrue(() -> server.locateQueue(multicastQueue).getMessageCount() == 0, 500, 50);
+ Wait.assertTrue(() -> server.locateQueue(divertMulticastQueue).getMessageCount() == 1, 2000, 100);
+
+ consumer.close();
+
+ consumer = session.createConsumer(anycastQueue);
+ session.start();
+ message = consumer.receive(1000);
+ assertNotNull(message);
+ message.acknowledge();
+ assertEquals(data + RoutingType.ANYCAST.toString(), message.getBodyBuffer().readString());
+ consumer.close();
+ Wait.assertTrue(() -> server.locateQueue(anycastQueue).getMessageCount() == 0, 500, 50);
+ Wait.assertTrue(() -> server.locateQueue(divertAnycastQueue).getMessageCount() == 1, 2000, 100);
+ }
+
+ @Test
+ public void testFilter() throws Exception {
+ final SimpleString queueName1 = SimpleString.toSimpleString("simpleQueue1");
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(10));
+ server.addAddressInfo(new AddressInfo(addressName));
+
+ ClientProducer producer = session.createProducer(addressName);
+ ClientMessage message = session.createMessage(false);
+ message.putLongProperty("xxx", 5);
+ producer.send(message);
+ message = session.createMessage(false);
+ message.putLongProperty("xxx", 15);
+ producer.send(message);
+ producer.close();
+ Wait.assertTrue(() -> server.locateQueue(divertQueue).getMessageCount() == 2, 500, 50);
+
+ session.createQueue(addressName.toString(), RoutingType.MULTICAST, queueName1.toString(), "xxx > 10", false);
+ Wait.assertTrue(() -> server.locateQueue(queueName1) != null);
+ Wait.assertTrue(() -> server.locateQueue(queueName1).getMessageCount() == 1, 500, 50);
+
+ ClientConsumer consumer = session.createConsumer(queueName1);
+ session.start();
+ message = consumer.receive(1000);
+ assertNotNull(message);
+ message.acknowledge();
+ assertEquals(15, (long) message.getLongProperty("xxx"));
+ consumer.close();
+ Wait.assertTrue(() -> server.locateQueue(queueName1).getMessageCount() == 0, 500, 50);
+ Wait.assertTrue(() -> server.locateQueue(divertQueue).getMessageCount() == 2, 2000, 100);
+ }
+
+ @Test
+ public void testAddressSettingOnRetroactiveResource() throws Exception {
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertAddress = ResourceNames.getRetroactiveResourceAddressName(internalNamingPrefix, delimiter, addressName);
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(10));
+ server.addAddressInfo(new AddressInfo(addressName));
+ assertEquals(-1, server.getAddressSettingsRepository().getMatch(divertAddress.toString()).getMaxSizeBytes());
+ server.getAddressSettingsRepository().addMatch("*" + delimiter + "*" + delimiter + "*" + delimiter + addressName + delimiter + "*" + delimiter + ResourceNames.RETROACTIVE_SUFFIX, new AddressSettings().setMaxSizeBytes(13));
+ assertEquals(13, server.getAddressSettingsRepository().getMatch(divertAddress.toString()).getMaxSizeBytes());
+ }
+
+ @Test
+ public void testPaging() throws Exception {
+ final SimpleString queueName = SimpleString.toSimpleString("simpleQueue");
+ final SimpleString randomQueueName = SimpleString.toSimpleString(UUID.randomUUID().toString());
+ final SimpleString addressName = SimpleString.toSimpleString("myAddress");
+ final SimpleString divertQueue = ResourceNames.getRetroactiveResourceQueueName(internalNamingPrefix, delimiter, addressName, RoutingType.MULTICAST);
+ final int MESSAGE_COUNT = 20;
+ final int MESSAGE_SIZE = 1024;
+
+ server.getAddressSettingsRepository().addMatch(addressName.toString(), new AddressSettings().setRetroactiveMessageCount(MESSAGE_COUNT).setMaxSizeBytes(1024 * 20).setPageSizeBytes(1024 * 10).setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE));
+ server.addAddressInfo(new AddressInfo(addressName));
+ server.createQueue(addressName, RoutingType.MULTICAST, randomQueueName, null, true, false);
+
+ ClientProducer producer = session.createProducer(addressName);
+
+ byte[] body = new byte[MESSAGE_SIZE];
+ ByteBuffer bb = ByteBuffer.wrap(body);
+ for (int j = 1; j <= MESSAGE_SIZE; j++) {
+ bb.put(getSamplebyte(j));
+ }
+
+ for (int i = 0; i < MESSAGE_COUNT * 2; i++) {
+ ClientMessage message = session.createMessage(true);
+ message.getBodyBuffer().writeBytes(body);
+ producer.send(message);
+ }
+ producer.close();
+ Wait.assertTrue(() -> server.locateQueue(randomQueueName).getMessageCount() == MESSAGE_COUNT * 2, 500, 50);
+
+ Wait.assertTrue(() -> server.locateQueue(divertQueue).getMessageCount() == MESSAGE_COUNT, 500, 50);
+
+ session.createQueue(addressName.toString(), RoutingType.MULTICAST, queueName.toString());
+ Wait.assertTrue(() -> server.locateQueue(queueName) != null);
+ Wait.assertTrue(() -> server.locateQueue(queueName).getMessageCount() == MESSAGE_COUNT, 500, 50);
+
+ ClientConsumer consumer = session.createConsumer(queueName);
+ session.start();
+
+ for (int i = 0; i < MESSAGE_COUNT; i++) {
+ ClientMessage message = consumer.receive(1000);
+ assertNotNull(message);
+ message.acknowledge();
+ }
+ consumer.close();
+ Wait.assertTrue(() -> server.locateQueue(queueName).getMessageCount() == 0, 500, 50);
+ Wait.assertTrue(() -> server.locateQueue(divertQueue).getMessageCount() == MESSAGE_COUNT, 2000, 100);
+ }
+}
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakeJournalLoader.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakeJournalLoader.java
index a84d2d0..3768b26 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakeJournalLoader.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakeJournalLoader.java
@@ -50,8 +50,7 @@ public class FakeJournalLoader implements JournalLoader {
}
@Override
- public void initAddresses(Map<Long, AddressBindingInfo> addressBindingInfosMap,
- List<AddressBindingInfo> addressBindingInfo) throws Exception {
+ public void initAddresses(List<AddressBindingInfo> addressBindingInfo) throws Exception {
}
@Override
[activemq-artemis] 03/03: This closes #2850
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit eb3c4931c2871a58fea9208eebaf90bc8ffbe462
Merge: 47a5406 84067d8
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Mon Oct 28 09:01:42 2019 -0400
This closes #2850
.../apache/activemq/artemis/logs/AuditLogger.java | 7 +
.../api/config/ActiveMQDefaultConfiguration.java | 6 +
.../api/core/management/ActiveMQServerControl.java | 58 ++-
.../api/core/management/AddressControl.java | 3 +
.../api/core/management/AddressSettingsInfo.java | 270 +++++++++++-
.../artemis/api/core/management/DivertControl.java | 3 +
.../artemis/api/core/management/QueueControl.java | 6 +
.../artemis/api/core/management/ResourceNames.java | 31 +-
.../core/management/AddressSettingsInfoTest.java | 57 ++-
.../api/core/management/ResourceNamesTest.java | 91 ++++
.../deployers/impl/FileConfigurationParser.java | 6 +
.../management/impl/ActiveMQServerControlImpl.java | 224 ++++++++--
.../core/management/impl/AddressControlImpl.java | 8 +
.../core/management/impl/DivertControlImpl.java | 15 +-
.../core/management/impl/QueueControlImpl.java | 16 +
.../core/postoffice/impl/PostOfficeImpl.java | 133 ++++++
.../core/postoffice/impl/SimpleAddressManager.java | 2 +-
.../artemis/core/server/RoutingContext.java | 2 +-
.../core/server/impl/ActiveMQServerImpl.java | 20 +-
.../artemis/core/server/impl/AddressInfo.java | 14 +-
.../artemis/core/server/impl/DivertImpl.java | 2 +-
.../artemis/core/server/impl/JournalLoader.java | 3 +-
.../core/server/impl/PostOfficeJournalLoader.java | 5 +-
.../artemis/core/server/impl/QueueImpl.java | 33 +-
.../core/server/impl/RoutingContextImpl.java | 3 +-
.../management/impl/ManagementServiceImpl.java | 2 +-
.../core/settings/impl/AddressSettings.java | 50 ++-
.../artemis/core/settings/impl/DeletionPolicy.java | 11 +
.../core/settings/impl/SlowConsumerPolicy.java | 11 +
.../resources/schema/artemis-configuration.xsd | 11 +-
.../core/config/impl/FileConfigurationTest.java | 2 +
.../impl/journal/AddressBindingEncodingTest.java | 52 +++
.../resources/ConfigurationTest-full-config.xml | 1 +
...rationTest-xinclude-config-address-settings.xml | 1 +
.../src/test/resources/artemis-configuration.xsd | 15 +
docs/user-manual/en/SUMMARY.md | 1 +
docs/user-manual/en/address-model.md | 13 +-
docs/user-manual/en/configuration-index.md | 4 +
docs/user-manual/en/retroactive-addresses.md | 88 ++++
.../RequestReplyMultiProtocolTest.java | 5 +-
.../crossprotocol/RequestReplyNonJMSTest.java | 14 +-
.../management/ActiveMQServerControlTest.java | 302 ++++++++++++-
.../ActiveMQServerControlUsingCoreTest.java | 155 ++++++-
.../integration/management/AddressControlTest.java | 13 +
.../management/AddressControlUsingCoreTest.java | 5 +
.../integration/management/DivertControlTest.java | 24 ++
.../management/DivertControlUsingCoreTest.java | 5 +
.../integration/management/QueueControlTest.java | 24 +-
.../management/QueueControlUsingCoreTest.java | 5 +
.../server/RetroactiveAddressFailoverTest.java | 119 ++++++
.../integration/server/RetroactiveAddressTest.java | 476 +++++++++++++++++++++
.../core/server/impl/fakes/FakeJournalLoader.java | 3 +-
52 files changed, 2341 insertions(+), 89 deletions(-)