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 2020/02/20 22:45:10 UTC
[activemq-artemis] 02/03: ARTEMIS-2624 auto-create expiry resources
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 d0758f34f64451865b2a69fa186650f35bc8383f
Author: Justin Bertram <jb...@apache.org>
AuthorDate: Fri Feb 14 12:39:19 2020 -0600
ARTEMIS-2624 auto-create expiry resources
---
.../api/core/management/ActiveMQServerControl.java | 5 +-
.../api/core/management/AddressSettingsInfo.java | 31 ++-
.../core/management/AddressSettingsInfoTest.java | 8 +-
.../deployers/impl/FileConfigurationParser.java | 12 ++
.../management/impl/ActiveMQServerControlImpl.java | 19 +-
.../artemis/core/server/impl/QueueImpl.java | 17 ++
.../core/settings/impl/AddressSettings.java | 101 +++++++++-
.../resources/schema/artemis-configuration.xsd | 25 +++
.../core/config/impl/FileConfigurationTest.java | 6 +
.../resources/ConfigurationTest-full-config.xml | 3 +
...rationTest-xinclude-config-address-settings.xml | 3 +
.../src/test/resources/artemis-configuration.xsd | 25 +++
docs/user-manual/en/address-model.md | 14 ++
docs/user-manual/en/message-expiry.md | 51 +++++
.../management/ActiveMQServerControlTest.java | 29 ++-
.../ActiveMQServerControlUsingCoreTest.java | 10 +-
.../server/AutoCreateExpiryResourcesTest.java | 212 +++++++++++++++++++++
17 files changed, 557 insertions(+), 14 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 dbd8e1a..d559526 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
@@ -1442,7 +1442,10 @@ public interface ActiveMQServerControl {
@Parameter(desc = "the number of messages to preserve for future queues created on the matching address", name = "retroactiveMessageCount") long retroactiveMessageCount,
@Parameter(desc = "allow dead-letter address & queue to be created automatically", name = "autoCreateDeadLetterResources") boolean autoCreateDeadLetterResources,
@Parameter(desc = "prefix to use on auto-create dead-letter queue", name = "deadLetterQueuePrefix") String deadLetterQueuePrefix,
- @Parameter(desc = "suffix to use on auto-create dead-letter queue", name = "deadLetterQueueSuffix") String deadLetterQueueSuffix) throws Exception;
+ @Parameter(desc = "suffix to use on auto-create dead-letter queue", name = "deadLetterQueueSuffix") String deadLetterQueueSuffix,
+ @Parameter(desc = "allow expiry address & queue to be created automatically", name = "autoCreateExpiryResources") boolean autoCreateExpiryResources,
+ @Parameter(desc = "prefix to use on auto-create expiry queue", name = "expiryQueuePrefix") String expiryQueuePrefix,
+ @Parameter(desc = "suffix to use on auto-create expiry queue", name = "expiryQueueSuffix") String expiryQueueSuffix) 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 5a570c2..5bd9037 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
@@ -123,6 +123,12 @@ public final class AddressSettingsInfo {
private final String deadLetterQueueSuffix;
+ private final boolean autoCreateExpiryResources;
+
+ private final String expiryQueuePrefix;
+
+ private final String expiryQueueSuffix;
+
// Static --------------------------------------------------------
public static AddressSettingsInfo from(final String jsonString) {
@@ -176,7 +182,10 @@ public final class AddressSettingsInfo {
object.getJsonNumber("retroactiveMessageCount").longValue(),
object.getBoolean("autoCreateDeadLetterResources"),
object.getString("deadLetterQueuePrefix"),
- object.getString("deadLetterQueueSuffix"));
+ object.getString("deadLetterQueueSuffix"),
+ object.getBoolean("autoCreateExpiryResources"),
+ object.getString("expiryQueuePrefix"),
+ object.getString("expiryQueueSuffix"));
}
// Constructors --------------------------------------------------
@@ -230,7 +239,10 @@ public final class AddressSettingsInfo {
long retroactiveMessageCount,
boolean autoCreateDeadLetterResources,
String deadLetterQueuePrefix,
- String deadLetterQueueSuffix) {
+ String deadLetterQueueSuffix,
+ boolean autoCreateExpiryResources,
+ String expiryQueuePrefix,
+ String expiryQueueSuffix) {
this.addressFullMessagePolicy = addressFullMessagePolicy;
this.maxSizeBytes = maxSizeBytes;
this.pageSizeBytes = pageSizeBytes;
@@ -281,6 +293,9 @@ public final class AddressSettingsInfo {
this.autoCreateDeadLetterResources = autoCreateDeadLetterResources;
this.deadLetterQueuePrefix = deadLetterQueuePrefix;
this.deadLetterQueueSuffix = deadLetterQueueSuffix;
+ this.autoCreateExpiryResources = autoCreateExpiryResources;
+ this.expiryQueuePrefix = expiryQueuePrefix;
+ this.expiryQueueSuffix = expiryQueueSuffix;
}
// Public --------------------------------------------------------
@@ -492,5 +507,17 @@ public final class AddressSettingsInfo {
public String getDeadLetterQueueSuffix() {
return deadLetterQueueSuffix;
}
+
+ public boolean isAutoCreateExpiryResources() {
+ return autoCreateExpiryResources;
+ }
+
+ public String getExpiryQueuePrefix() {
+ return expiryQueuePrefix;
+ }
+
+ public String getExpiryQueueSuffix() {
+ return expiryQueueSuffix;
+ }
}
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 aaa9595..4aee4d8 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
@@ -80,7 +80,10 @@ public class AddressSettingsInfoTest {
"\"retroactiveMessageCount\":101,\n" +
"\"autoCreateDeadLetterResources\":true,\n" +
"\"deadLetterQueuePrefix\":\"FOO.\",\n" +
- "\"deadLetterQueueSuffix\":\".FOO\"\n" +
+ "\"deadLetterQueueSuffix\":\".FOO\",\n" +
+ "\"autoCreateExpiryResources\":true,\n" +
+ "\"expiryQueuePrefix\":\"BAR.\",\n" +
+ "\"expiryQueueSuffix\":\".BAR\"\n" +
"}";
AddressSettingsInfo addressSettingsInfo = AddressSettingsInfo.from(json);
assertEquals("fullPolicy", addressSettingsInfo.getAddressFullMessagePolicy());
@@ -133,6 +136,9 @@ public class AddressSettingsInfoTest {
assertTrue(addressSettingsInfo.isAutoCreateDeadLetterResources());
assertEquals("FOO.", addressSettingsInfo.getDeadLetterQueuePrefix());
assertEquals(".FOO", addressSettingsInfo.getDeadLetterQueueSuffix());
+ assertTrue(addressSettingsInfo.isAutoCreateExpiryResources());
+ assertEquals("BAR.", addressSettingsInfo.getExpiryQueuePrefix());
+ assertEquals(".BAR", addressSettingsInfo.getExpiryQueueSuffix());
}
}
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 8305ef6..67eeb15 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
@@ -169,6 +169,12 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
private static final String EXPIRY_ADDRESS_NODE_NAME = "expiry-address";
+ private static final String AUTO_CREATE_EXPIRY_RESOURCES_NODE_NAME = "auto-create-expiry-resources";
+
+ private static final String EXPIRY_QUEUE_PREFIX_NODE_NAME = "expiry-queue-prefix";
+
+ private static final String EXPIRY_QUEUE_SUFFIX_NODE_NAME = "expiry-queue-suffix";
+
private static final String EXPIRY_DELAY_NODE_NAME = "expiry-delay";
private static final String REDELIVERY_DELAY_NODE_NAME = "redelivery-delay";
@@ -1197,6 +1203,12 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
addressSettings.setDeadLetterQueuePrefix(new SimpleString(getTrimmedTextContent(child)));
} else if (DEAD_LETTER_QUEUE_SUFFIX_NODE_NAME.equalsIgnoreCase(name)) {
addressSettings.setDeadLetterQueueSuffix(new SimpleString(getTrimmedTextContent(child)));
+ } else if (AUTO_CREATE_EXPIRY_RESOURCES_NODE_NAME.equalsIgnoreCase(name)) {
+ addressSettings.setAutoCreateExpiryResources(XMLUtil.parseBoolean(child));
+ } else if (EXPIRY_QUEUE_PREFIX_NODE_NAME.equalsIgnoreCase(name)) {
+ addressSettings.setExpiryQueuePrefix(new SimpleString(getTrimmedTextContent(child)));
+ } else if (EXPIRY_QUEUE_SUFFIX_NODE_NAME.equalsIgnoreCase(name)) {
+ addressSettings.setExpiryQueueSuffix(new SimpleString(getTrimmedTextContent(child)));
}
}
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 ecc5d25..cb57760 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
@@ -2756,6 +2756,9 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
.add("autoCreateDeadLetterResources", addressSettings.isAutoCreateDeadLetterResources())
.add("deadLetterQueuePrefix", addressSettings.getDeadLetterQueuePrefix().toString())
.add("deadLetterQueueSuffix", addressSettings.getDeadLetterQueueSuffix().toString())
+ .add("autoCreateExpiryResources", addressSettings.isAutoCreateExpiryResources())
+ .add("expiryQueuePrefix", addressSettings.getExpiryQueuePrefix().toString())
+ .add("expiryQueueSuffix", addressSettings.getExpiryQueueSuffix().toString())
.build()
.toString();
}
@@ -2990,7 +2993,10 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
retroactiveMessageCount,
AddressSettings.DEFAULT_AUTO_CREATE_DEAD_LETTER_RESOURCES,
AddressSettings.DEFAULT_DEAD_LETTER_QUEUE_PREFIX.toString(),
- AddressSettings.DEFAULT_DEAD_LETTER_QUEUE_SUFFIX.toString());
+ AddressSettings.DEFAULT_DEAD_LETTER_QUEUE_SUFFIX.toString(),
+ AddressSettings.DEFAULT_AUTO_CREATE_EXPIRY_RESOURCES,
+ AddressSettings.DEFAULT_EXPIRY_QUEUE_PREFIX.toString(),
+ AddressSettings.DEFAULT_EXPIRY_QUEUE_SUFFIX.toString());
}
@Override
@@ -3045,7 +3051,10 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
final long retroactiveMessageCount,
final boolean autoCreateDeadLetterResources,
final String deadLetterQueuePrefix,
- final String deadLetterQueueSuffix) throws Exception {
+ final String deadLetterQueueSuffix,
+ final boolean autoCreateExpiryResources,
+ final String expiryQueuePrefix,
+ final String expiryQueueSuffix) throws Exception {
if (AuditLogger.isEnabled()) {
AuditLogger.addAddressSettings(this.server, address, DLA, expiryAddress, expiryDelay, defaultLastValueQueue, maxDeliveryAttempts,
maxSizeBytes, pageSizeBytes, pageMaxCacheSize, redeliveryDelay, redeliveryMultiplier,
@@ -3058,7 +3067,8 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
defaultDelayBeforeDispatch, defaultQueueRoutingType, defaultAddressRoutingType, defaultConsumerWindowSize,
defaultRingSize, autoDeleteCreatedQueues, autoDeleteQueuesDelay, autoDeleteQueuesMessageCount,
autoDeleteAddressesDelay, redeliveryCollisionAvoidanceFactor, retroactiveMessageCount, autoCreateDeadLetterResources,
- deadLetterQueuePrefix, deadLetterQueueSuffix);
+ deadLetterQueuePrefix, deadLetterQueueSuffix, autoCreateExpiryResources, expiryQueuePrefix,
+ expiryQueueSuffix);
}
checkStarted();
@@ -3123,6 +3133,9 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
addressSettings.setAutoCreateDeadLetterResources(autoCreateDeadLetterResources);
addressSettings.setDeadLetterQueuePrefix(deadLetterQueuePrefix == null ? null : new SimpleString(deadLetterQueuePrefix));
addressSettings.setDeadLetterQueueSuffix(deadLetterQueueSuffix == null ? null : new SimpleString(deadLetterQueueSuffix));
+ addressSettings.setAutoCreateExpiryResources(autoCreateExpiryResources);
+ addressSettings.setExpiryQueuePrefix(expiryQueuePrefix == null ? null : new SimpleString(expiryQueuePrefix));
+ addressSettings.setExpiryQueueSuffix(expiryQueueSuffix == null ? null : new SimpleString(expiryQueueSuffix));
server.getAddressSettingsRepository().addMatch(address, addressSettings);
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 13bcd4f..b21e1a2 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
@@ -1793,6 +1793,9 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
}
if (messageExpiryAddress != null) {
+
+ createExpiryResources();
+
if (logger.isTraceEnabled()) {
logger.trace("moving expired reference " + ref + " to address = " + messageExpiryAddress + " from queue=" + this.getName());
}
@@ -3279,6 +3282,9 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
SimpleString expiryAddress = addressSettingsRepository.getMatch(address.toString()).getExpiryAddress();
if (expiryAddress != null) {
+
+ createExpiryResources();
+
Bindings bindingList = postOffice.lookupBindingsForAddress(expiryAddress);
if (bindingList == null || bindingList.getBindings().isEmpty()) {
@@ -3373,6 +3379,17 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
}
}
+ private void createExpiryResources() throws Exception {
+ AddressSettings addressSettings = server.getAddressSettingsRepository().getMatch(getAddress().toString());
+ if (addressSettings.isAutoCreateExpiryResources() && !getAddress().equals(addressSettings.getExpiryAddress())) {
+ if (addressSettings.getExpiryAddress() != null && addressSettings.getExpiryAddress().length() != 0) {
+ SimpleString expiryQueueName = addressSettings.getExpiryQueuePrefix().concat(getAddress()).concat(addressSettings.getExpiryQueueSuffix());
+ SimpleString expiryFilter = new SimpleString(String.format("%s = '%s'", Message.HDR_ORIGINAL_ADDRESS, getAddress()));
+ server.createQueue(addressSettings.getExpiryAddress(), RoutingType.MULTICAST, expiryQueueName, expiryFilter, null, true, false, true, false, true, addressSettings.getDefaultMaxConsumers(), addressSettings.isDefaultPurgeOnNoConsumers(), addressSettings.isDefaultExclusiveQueue(), addressSettings.isDefaultLastValueQueue(), true);
+ }
+ }
+ }
+
private void move(final Transaction originalTX,
final SimpleString address,
final Binding binding,
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 b152a39..1eb8121 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
@@ -92,6 +92,12 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
public static final long DEFAULT_REDISTRIBUTION_DELAY = -1;
+ public static final boolean DEFAULT_AUTO_CREATE_EXPIRY_RESOURCES = false;
+
+ public static final SimpleString DEFAULT_EXPIRY_QUEUE_PREFIX = SimpleString.toSimpleString("EXP.");
+
+ public static final SimpleString DEFAULT_EXPIRY_QUEUE_SUFFIX = SimpleString.toSimpleString("");
+
public static final long DEFAULT_EXPIRY_DELAY = -1;
public static final boolean DEFAULT_SEND_TO_DLA_ON_NO_ROUTE = false;
@@ -227,6 +233,12 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
private SimpleString deadLetterQueueSuffix = null;
+ private Boolean autoCreateExpiryResources = null;
+
+ private SimpleString expiryQueuePrefix = null;
+
+ private SimpleString expiryQueueSuffix = null;
+
//from amq5
//make it transient
private transient Integer queuePrefetch = null;
@@ -248,6 +260,9 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
this.deadLetterQueuePrefix = other.deadLetterQueuePrefix;
this.deadLetterQueueSuffix = other.deadLetterQueueSuffix;
this.expiryAddress = other.expiryAddress;
+ this.autoCreateExpiryResources = other.autoCreateExpiryResources;
+ this.expiryQueuePrefix = other.expiryQueuePrefix;
+ this.expiryQueueSuffix = other.expiryQueueSuffix;
this.expiryDelay = other.expiryDelay;
this.defaultLastValueQueue = other.defaultLastValueQueue;
this.defaultLastValueKey = other.defaultLastValueKey;
@@ -626,6 +641,33 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
return this;
}
+ public boolean isAutoCreateExpiryResources() {
+ return autoCreateExpiryResources != null ? autoCreateExpiryResources : AddressSettings.DEFAULT_AUTO_CREATE_EXPIRY_RESOURCES;
+ }
+
+ public AddressSettings setAutoCreateExpiryResources(final boolean value) {
+ autoCreateExpiryResources = value;
+ return this;
+ }
+
+ public SimpleString getExpiryQueuePrefix() {
+ return expiryQueuePrefix != null ? expiryQueuePrefix : AddressSettings.DEFAULT_EXPIRY_QUEUE_PREFIX;
+ }
+
+ public AddressSettings setExpiryQueuePrefix(final SimpleString value) {
+ expiryQueuePrefix = value;
+ return this;
+ }
+
+ public SimpleString getExpiryQueueSuffix() {
+ return expiryQueueSuffix != null ? expiryQueueSuffix : AddressSettings.DEFAULT_EXPIRY_QUEUE_SUFFIX;
+ }
+
+ public AddressSettings setExpiryQueueSuffix(final SimpleString value) {
+ expiryQueueSuffix = value;
+ return this;
+ }
+
public Long getExpiryDelay() {
return expiryDelay;
}
@@ -984,6 +1026,15 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
if (deadLetterQueueSuffix == null) {
deadLetterQueueSuffix = merged.deadLetterQueueSuffix;
}
+ if (autoCreateExpiryResources == null) {
+ autoCreateExpiryResources = merged.autoCreateExpiryResources;
+ }
+ if (expiryQueuePrefix == null) {
+ expiryQueuePrefix = merged.expiryQueuePrefix;
+ }
+ if (expiryQueueSuffix == null) {
+ expiryQueueSuffix = merged.expiryQueueSuffix;
+ }
}
@Override
@@ -1168,6 +1219,18 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
if (buffer.readableBytes() > 0) {
deadLetterQueueSuffix = buffer.readNullableSimpleString();
}
+
+ if (buffer.readableBytes() > 0) {
+ autoCreateExpiryResources = BufferHelper.readNullableBoolean(buffer);
+ }
+
+ if (buffer.readableBytes() > 0) {
+ expiryQueuePrefix = buffer.readNullableSimpleString();
+ }
+
+ if (buffer.readableBytes() > 0) {
+ expiryQueueSuffix = buffer.readNullableSimpleString();
+ }
}
@Override
@@ -1224,7 +1287,10 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
BufferHelper.sizeOfNullableLong(retroactiveMessageCount) +
BufferHelper.sizeOfNullableBoolean(autoCreateDeadLetterResources) +
SimpleString.sizeofNullableString(deadLetterQueuePrefix) +
- SimpleString.sizeofNullableString(deadLetterQueueSuffix);
+ SimpleString.sizeofNullableString(deadLetterQueueSuffix) +
+ BufferHelper.sizeOfNullableBoolean(autoCreateExpiryResources) +
+ SimpleString.sizeofNullableString(expiryQueuePrefix) +
+ SimpleString.sizeofNullableString(expiryQueueSuffix);
}
@Override
@@ -1336,6 +1402,12 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
buffer.writeNullableSimpleString(deadLetterQueuePrefix);
buffer.writeNullableSimpleString(deadLetterQueueSuffix);
+
+ BufferHelper.writeNullableBoolean(buffer, autoCreateExpiryResources);
+
+ buffer.writeNullableSimpleString(expiryQueuePrefix);
+
+ buffer.writeNullableSimpleString(expiryQueueSuffix);
}
/* (non-Javadoc)
@@ -1400,6 +1472,9 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
result = prime * result + ((autoCreateDeadLetterResources == null) ? 0 : autoCreateDeadLetterResources.hashCode());
result = prime * result + ((deadLetterQueuePrefix == null) ? 0 : deadLetterQueuePrefix.hashCode());
result = prime * result + ((deadLetterQueueSuffix == null) ? 0 : deadLetterQueueSuffix.hashCode());
+ result = prime * result + ((autoCreateExpiryResources == null) ? 0 : autoCreateExpiryResources.hashCode());
+ result = prime * result + ((expiryQueuePrefix == null) ? 0 : expiryQueuePrefix.hashCode());
+ result = prime * result + ((expiryQueueSuffix == null) ? 0 : expiryQueueSuffix.hashCode());
return result;
}
@@ -1707,6 +1782,24 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
} else if (!deadLetterQueueSuffix.equals(other.deadLetterQueueSuffix))
return false;
+ if (autoCreateExpiryResources == null) {
+ if (other.autoCreateExpiryResources != null)
+ return false;
+ } else if (!autoCreateExpiryResources.equals(other.autoCreateExpiryResources))
+ return false;
+
+ if (expiryQueuePrefix == null) {
+ if (other.expiryQueuePrefix != null)
+ return false;
+ } else if (!expiryQueuePrefix.equals(other.expiryQueuePrefix))
+ return false;
+
+ if (expiryQueueSuffix == null) {
+ if (other.expiryQueueSuffix != null)
+ return false;
+ } else if (!expiryQueueSuffix.equals(other.expiryQueueSuffix))
+ return false;
+
return true;
}
@@ -1822,6 +1915,12 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
deadLetterQueuePrefix +
", deadLetterQueueSuffix=" +
deadLetterQueueSuffix +
+ ", autoCreateExpiryResources=" +
+ autoCreateExpiryResources +
+ ", expiryQueuePrefix=" +
+ expiryQueuePrefix +
+ ", expiryQueueSuffix=" +
+ expiryQueueSuffix +
"]";
}
}
diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
index 867e0dd..2b7b1ca 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -3102,6 +3102,31 @@
</xsd:annotation>
</xsd:element>
+ <xsd:element name="auto-create-expiry-resources" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ whether or not to automatically create the expiry-address and/or a corresponding queue
+ on that address when a message is sent to a matching queue
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="expiry-queue-prefix" type="xsd:string" default="EXP." maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the prefix to use for auto-created expiry queues
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="expiry-queue-suffix" type="xsd:string" default="" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the suffix to use for auto-created expiry queues
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
<xsd:element name="expiry-delay" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
<xsd:annotation>
<xsd:documentation>
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 d90c0fb..c0046b8 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
@@ -346,6 +346,9 @@ public class FileConfigurationTest extends ConfigurationImplTest {
assertEquals(AddressSettings.DEFAULT_DEAD_LETTER_QUEUE_PREFIX, conf.getAddressesSettings().get("a1").getDeadLetterQueuePrefix());
assertEquals(AddressSettings.DEFAULT_DEAD_LETTER_QUEUE_SUFFIX, conf.getAddressesSettings().get("a1").getDeadLetterQueueSuffix());
assertEquals("a1.2", conf.getAddressesSettings().get("a1").getExpiryAddress().toString());
+ assertEquals(AddressSettings.DEFAULT_AUTO_CREATE_EXPIRY_RESOURCES, conf.getAddressesSettings().get("a1").isAutoCreateExpiryResources());
+ assertEquals(AddressSettings.DEFAULT_EXPIRY_QUEUE_PREFIX, conf.getAddressesSettings().get("a1").getExpiryQueuePrefix());
+ assertEquals(AddressSettings.DEFAULT_EXPIRY_QUEUE_SUFFIX, conf.getAddressesSettings().get("a1").getExpiryQueueSuffix());
assertEquals(1, conf.getAddressesSettings().get("a1").getRedeliveryDelay());
assertEquals(0.5, conf.getAddressesSettings().get("a1").getRedeliveryCollisionAvoidanceFactor(), 0);
assertEquals(856686592L, conf.getAddressesSettings().get("a1").getMaxSizeBytes());
@@ -373,6 +376,9 @@ public class FileConfigurationTest extends ConfigurationImplTest {
assertEquals("", conf.getAddressesSettings().get("a2").getDeadLetterQueuePrefix().toString());
assertEquals(".DLQ", conf.getAddressesSettings().get("a2").getDeadLetterQueueSuffix().toString());
assertEquals("a2.2", conf.getAddressesSettings().get("a2").getExpiryAddress().toString());
+ assertEquals(true, conf.getAddressesSettings().get("a2").isAutoCreateDeadLetterResources());
+ assertEquals("", conf.getAddressesSettings().get("a2").getExpiryQueuePrefix().toString());
+ assertEquals(".EXP", conf.getAddressesSettings().get("a2").getExpiryQueueSuffix().toString());
assertEquals(5, conf.getAddressesSettings().get("a2").getRedeliveryDelay());
assertEquals(0.0, conf.getAddressesSettings().get("a2").getRedeliveryCollisionAvoidanceFactor(), 0);
assertEquals(932489234928324L, conf.getAddressesSettings().get("a2").getMaxSizeBytes());
diff --git a/artemis-server/src/test/resources/ConfigurationTest-full-config.xml b/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
index 8ecebc1..c54d38f 100644
--- a/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
+++ b/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
@@ -442,6 +442,9 @@
<dead-letter-queue-prefix></dead-letter-queue-prefix>
<dead-letter-queue-suffix>.DLQ</dead-letter-queue-suffix>
<expiry-address>a2.2</expiry-address>
+ <auto-create-expiry-resources>true</auto-create-expiry-resources>
+ <expiry-queue-prefix></expiry-queue-prefix>
+ <expiry-queue-suffix>.EXP</expiry-queue-suffix>
<redelivery-delay>5</redelivery-delay>
<max-size-bytes>932489234928324</max-size-bytes>
<page-size-bytes>712671626</page-size-bytes>
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 5ff60da..8ba173a 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
@@ -47,6 +47,9 @@
<dead-letter-queue-prefix></dead-letter-queue-prefix>
<dead-letter-queue-suffix>.DLQ</dead-letter-queue-suffix>
<expiry-address>a2.2</expiry-address>
+ <auto-create-expiry-resources>true</auto-create-expiry-resources>
+ <expiry-queue-prefix></expiry-queue-prefix>
+ <expiry-queue-suffix>.EXP</expiry-queue-suffix>
<redelivery-delay>5</redelivery-delay>
<max-size-bytes>932489234928324</max-size-bytes>
<page-size-bytes>712671626</page-size-bytes>
diff --git a/artemis-tools/src/test/resources/artemis-configuration.xsd b/artemis-tools/src/test/resources/artemis-configuration.xsd
index dce2fbc..280098e 100644
--- a/artemis-tools/src/test/resources/artemis-configuration.xsd
+++ b/artemis-tools/src/test/resources/artemis-configuration.xsd
@@ -2958,6 +2958,31 @@
</xsd:annotation>
</xsd:element>
+ <xsd:element name="auto-create-expiry-resources" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ whether or not to automatically create the expiry-address and/or a corresponding queue
+ on that address when a message is sent to a matching queue
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="expiry-queue-prefix" type="xsd:string" default="EXP." maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the prefix to use for auto-created expiry queues
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
+ <xsd:element name="expiry-queue-suffix" type="xsd:string" default="" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ the suffix to use for auto-created expiry queues
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
<xsd:element name="expiry-delay" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
<xsd:annotation>
<xsd:documentation>
diff --git a/docs/user-manual/en/address-model.md b/docs/user-manual/en/address-model.md
index 33f4d1f..97f9e0e 100644
--- a/docs/user-manual/en/address-model.md
+++ b/docs/user-manual/en/address-model.md
@@ -573,6 +573,9 @@ that would be found in the `broker.xml` file.
<dead-letter-queue-prefix>DLQ.</dead-letter-queue-prefix>
<dead-letter-queue-suffix></dead-letter-queue-suffix>
<expiry-address>ExpiryQueue</expiry-address>
+ <auto-create-expiry-resources>false</auto-create-expiry-resources>
+ <expiry-queue-prefix>EXP.</expiry-queue-prefix>
+ <expiry-queue-suffix></expiry-queue-suffix>
<expiry-delay>123</expiry-delay>
<redelivery-delay>5000</redelivery-delay>
<redelivery-delay-multiplier>1.0</redelivery-delay-multiplier>
@@ -656,6 +659,17 @@ dead-letter queues. Read more in the chapter about
address is defined here then such messages will simply be discarded. Read more
about [message expiry](message-expiry.md#configuring-expiry-addresses).
+`auto-create-expiry-resources` determines whether or not the broker will
+automatically create the defined `expiry-address` and a corresponding expiry
+queue when a message expired. Read more in the chapter about
+[undelivered messages](undelivered-messages.md).
+
+`expiry-queue-prefix` defines the prefix used for automatically created expiry
+queues. Read more in the chapter about [message expiry](message-expiry.md).
+
+`expiry-queue-suffix` defines the suffix used for automatically created expiry
+queues. Read more in the chapter about [message expiry](message-expiry.md).
+
`expiry-delay` defines the expiration time that will be used for messages which
are using the default expiration time (i.e. 0). For example, if `expiry-delay`
is set to "10" and a message which is using the default expiration time (i.e.
diff --git a/docs/user-manual/en/message-expiry.md b/docs/user-manual/en/message-expiry.md
index b8b29e7..e512b2d 100644
--- a/docs/user-manual/en/message-expiry.md
+++ b/docs/user-manual/en/message-expiry.md
@@ -53,6 +53,7 @@ Default Expiry delay can be configured in the address-setting configuration:
```xml
<!-- expired messages in exampleQueue will be sent to the expiry address expiryQueue -->
<address-setting match="exampleQueue">
+ <expiry-address>expiryQueue</expiry-address>
<expiry-delay>10</expiry-delay>
</address-setting>
```
@@ -82,6 +83,56 @@ If messages are expired and no expiry address is specified, messages are simply
removed from the queue and dropped. Address [wildcards](wildcard-syntax.md) can
be used to configure expiry address for a set of addresses.
+## Configuring Automatic Creation of Expiry Resources
+
+It's common to segregate expired messages by their original address.
+For example, a message sent to the `stocks` address that expired for some
+reason might be ultimately routed to the `EXP.stocks` queue, and likewise
+a message sent to the `orders` address that expired might be routed to
+the `EXP.orders` queue.
+
+Using this pattern can make it easy to track and administrate
+expired messages. However, it can pose a challenge in environments
+which predominantly use auto-created addresses and queues. Typically
+administrators in those environments don't want to manually create
+an `address-setting` to configure the `expiry-address` much less
+the actual `address` and `queue` to hold the expired messages.
+
+The solution to this problem is to set the `auto-create-expiry-resources`
+`address-setting` to `true` (it's `false` by default) so that the
+broker will create the `address` and `queue` to deal with the
+expired messages automatically. The `address` created will be the
+one defined by the `expiry-address`. A `MULTICAST` `queue` will be
+created on that `address`. It will be named by the `address` to which
+the message was originally sent, and it will have a filter defined using
+the aforementioned `_AMQ_ORIG_ADDRESS` property so that it will only
+receive messages sent to the relevant `address`. The `queue` name can be
+configured with a prefix and suffix. See the relevant settings in the
+table below:
+
+`address-setting`|default
+---|---
+`expiry-queue-prefix`|`EXP.`
+`expiry-queue-suffix`|`` (empty string)
+
+Here is an example configuration:
+
+```xml
+<address-setting match="#">
+ <expiry-address>expiryAddress</expiry-address>
+ <auto-create-expiry-resources>true</auto-create-expiry-resources>
+ <expiry-queue-prefix></expiry-queue-prefix> <!-- override the default -->
+ <expiry-queue-suffix>.EXP</expiry-queue-suffix>
+</address-setting>
+```
+
+The queue holding the expired messages can be accessed directly
+either by using the queue's name by itself (e.g. when using the core
+client) or by using the fully qualified queue name (e.g. when using
+a JMS client) just like any other queue. Also, note that the queue is
+auto-created which means it will be auto-deleted as per the relevant
+`address-settings`.
+
## Configuring The Expiry Reaper Thread
A reaper thread will periodically inspect the queues to check if messages have
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 74c94e8..910ea14 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
@@ -757,6 +757,9 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
boolean autoCreateDeadLetterResources = RandomUtil.randomBoolean();
String deadLetterQueuePrefix = RandomUtil.randomString();
String deadLetterQueueSuffix = RandomUtil.randomString();
+ boolean autoCreateExpiryResources = RandomUtil.randomBoolean();
+ String expiryQueuePrefix = RandomUtil.randomString();
+ String expiryQueueSuffix = RandomUtil.randomString();
serverControl.addAddressSettings(addressMatch,
DLA,
@@ -809,7 +812,10 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
retroactiveMessageCount,
autoCreateDeadLetterResources,
deadLetterQueuePrefix,
- deadLetterQueueSuffix);
+ deadLetterQueueSuffix,
+ autoCreateExpiryResources,
+ expiryQueuePrefix,
+ expiryQueueSuffix);
boolean ex = false;
try {
@@ -864,7 +870,10 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
retroactiveMessageCount,
autoCreateDeadLetterResources,
deadLetterQueuePrefix,
- deadLetterQueueSuffix);
+ deadLetterQueueSuffix,
+ autoCreateExpiryResources,
+ expiryQueuePrefix,
+ expiryQueueSuffix);
} catch (Exception expected) {
ex = true;
}
@@ -926,6 +935,9 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
assertEquals(autoCreateDeadLetterResources, info.isAutoCreateDeadLetterResources());
assertEquals(deadLetterQueuePrefix, info.getDeadLetterQueuePrefix());
assertEquals(deadLetterQueueSuffix, info.getDeadLetterQueueSuffix());
+ assertEquals(autoCreateExpiryResources, info.isAutoCreateExpiryResources());
+ assertEquals(expiryQueuePrefix, info.getExpiryQueuePrefix());
+ assertEquals(expiryQueueSuffix, info.getExpiryQueueSuffix());
serverControl.addAddressSettings(addressMatch,
DLA,
@@ -978,7 +990,10 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
retroactiveMessageCount,
autoCreateDeadLetterResources,
deadLetterQueuePrefix,
- deadLetterQueueSuffix);
+ deadLetterQueueSuffix,
+ autoCreateExpiryResources,
+ expiryQueuePrefix,
+ expiryQueueSuffix);
jsonString = serverControl.getAddressSettingsAsJSON(exactAddress);
info = AddressSettingsInfo.from(jsonString);
@@ -1033,6 +1048,9 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
assertEquals(autoCreateDeadLetterResources, info.isAutoCreateDeadLetterResources());
assertEquals(deadLetterQueuePrefix, info.getDeadLetterQueuePrefix());
assertEquals(deadLetterQueueSuffix, info.getDeadLetterQueueSuffix());
+ assertEquals(autoCreateExpiryResources, info.isAutoCreateExpiryResources());
+ assertEquals(expiryQueuePrefix, info.getExpiryQueuePrefix());
+ assertEquals(expiryQueueSuffix, info.getExpiryQueueSuffix());
ex = false;
try {
@@ -1087,7 +1105,10 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
retroactiveMessageCount,
autoCreateDeadLetterResources,
deadLetterQueuePrefix,
- deadLetterQueueSuffix);
+ deadLetterQueueSuffix,
+ autoCreateExpiryResources,
+ expiryQueuePrefix,
+ expiryQueueSuffix);
} 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 e18acf6..4bf0e8b 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
@@ -1060,7 +1060,10 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
@Parameter(desc = "the number of messages to preserve for future queues created on the matching address", name = "retroactiveMessageCount") long retroactiveMessageCount,
@Parameter(desc = "allow dead-letter address & queue to be created automatically", name = "autoCreateDeadLetterResources") boolean autoCreateDeadLetterResources,
@Parameter(desc = "prefix to use on auto-create dead-letter queue", name = "deadLetterQueuePrefix") String deadLetterQueuePrefix,
- @Parameter(desc = "suffix to use on auto-create dead-letter queue", name = "deadLetterQueueSuffix") String deadLetterQueueSuffix) throws Exception {
+ @Parameter(desc = "suffix to use on auto-create dead-letter queue", name = "deadLetterQueueSuffix") String deadLetterQueueSuffix,
+ @Parameter(desc = "allow expiry address & queue to be created automatically", name = "autoCreateExpiryResources") boolean autoCreateExpiryResources,
+ @Parameter(desc = "prefix to use on auto-create expiry queue", name = "expiryQueuePrefix") String expiryQueuePrefix,
+ @Parameter(desc = "suffix to use on auto-create expiry queue", name = "expiryQueueSuffix") String expiryQueueSuffix) throws Exception {
proxy.invokeOperation("addAddressSettings",
addressMatch,
DLA,
@@ -1113,7 +1116,10 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
retroactiveMessageCount,
autoCreateDeadLetterResources,
deadLetterQueuePrefix,
- deadLetterQueueSuffix);
+ deadLetterQueueSuffix,
+ autoCreateExpiryResources,
+ expiryQueuePrefix,
+ expiryQueueSuffix);
}
@Override
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateExpiryResourcesTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateExpiryResourcesTest.java
new file mode 100644
index 0000000..6800435
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateExpiryResourcesTest.java
@@ -0,0 +1,212 @@
+/**
+ * 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 javax.jms.JMSContext;
+
+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.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.Queue;
+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.RandomUtil;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.apache.activemq.artemis.utils.CompositeAddress;
+import org.junit.Before;
+import org.junit.Test;
+
+public class AutoCreateExpiryResourcesTest extends ActiveMQTestBase {
+ public final SimpleString addressA = new SimpleString("addressA");
+ public final SimpleString queueA = new SimpleString("queueA");
+ public final SimpleString expiryAddress = new SimpleString("myExpiry");
+
+ private ActiveMQServer server;
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ server = createServer(false);
+
+ // set common address settings needed for all tests; make sure to use getMatch instead of addMatch in invidual tests or these will be overwritten
+ server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoCreateExpiryResources(true).setExpiryAddress(expiryAddress).setExpiryDelay(100L));
+ server.getConfiguration().setMessageExpiryScanPeriod(50L);
+
+ server.start();
+ }
+
+ @Test
+ public void testAutoCreationOfExpiryResources() throws Exception {
+ int before = server.getActiveMQServerControl().getQueueNames().length;
+ triggerExpiration();
+ Wait.assertTrue(() -> server.getAddressInfo(expiryAddress) != null, 2000, 100);
+ assertNotNull(server.locateQueue(getDefaultExpiryQueueName(addressA)));
+ assertEquals(2, server.getActiveMQServerControl().getQueueNames().length - before);
+ }
+
+ @Test
+ public void testAutoCreationOfExpiryResourcesWithNullExpiry() throws Exception {
+ testAutoCreationOfExpiryResourcesWithNoExpiry(null);
+ }
+
+ @Test
+ public void testAutoCreationOfExpiryResourcesWithEmptyExpiry() throws Exception {
+ testAutoCreationOfExpiryResourcesWithNoExpiry(SimpleString.toSimpleString(""));
+ }
+
+ private void testAutoCreationOfExpiryResourcesWithNoExpiry(SimpleString expiryAddress) throws Exception {
+ server.getAddressSettingsRepository().getMatch("#").setExpiryAddress(expiryAddress);
+ int before = server.getActiveMQServerControl().getQueueNames().length;
+ triggerExpiration();
+ if (expiryAddress != null) {
+ assertNull(server.getAddressInfo(expiryAddress));
+ }
+ assertNull(server.locateQueue(getDefaultExpiryQueueName(addressA)));
+ assertEquals(1, server.getActiveMQServerControl().getQueueNames().length - before);
+ }
+
+ @Test
+ public void testAutoCreateExpiryQueuePrefix() throws Exception {
+ SimpleString prefix = RandomUtil.randomSimpleString();
+ server.getAddressSettingsRepository().getMatch("#").setExpiryQueuePrefix(prefix);
+ triggerExpiration();
+ Wait.assertTrue(() -> server.locateQueue(prefix.concat(addressA).concat(AddressSettings.DEFAULT_EXPIRY_QUEUE_SUFFIX)) != null, 2000, 100);
+ }
+
+ @Test
+ public void testAutoCreateExpiryQueueSuffix() throws Exception {
+ SimpleString suffix = RandomUtil.randomSimpleString();
+ server.getAddressSettingsRepository().getMatch("#").setExpiryQueueSuffix(suffix);
+ triggerExpiration();
+ Wait.assertTrue(() -> server.locateQueue(AddressSettings.DEFAULT_EXPIRY_QUEUE_PREFIX.concat(addressA).concat(suffix)) != null, 2000, 100);
+ }
+
+ @Test
+ public void testAutoCreateExpiryQueuePrefixAndSuffix() throws Exception {
+ SimpleString prefix = RandomUtil.randomSimpleString();
+ SimpleString suffix = RandomUtil.randomSimpleString();
+ server.getAddressSettingsRepository().getMatch("#").setExpiryQueuePrefix(prefix).setExpiryQueueSuffix(suffix);
+ triggerExpiration();
+ Wait.assertTrue(() -> server.locateQueue(prefix.concat(addressA).concat(suffix)) != null, 2000, 100);
+ }
+
+ @Test
+ public void testAutoCreatedExpiryFilterAnycast() throws Exception {
+ testAutoCreatedExpiryFilter(RoutingType.ANYCAST);
+ }
+
+ @Test
+ public void testAutoCreatedExpiryFilterMulticast() throws Exception {
+ testAutoCreatedExpiryFilter(RoutingType.MULTICAST);
+ }
+
+ private void testAutoCreatedExpiryFilter(RoutingType routingType) throws Exception {
+ final int ITERATIONS = 50;
+ final int MESSAGE_COUNT = 5;
+
+ for (int i = 0; i < ITERATIONS; i++) {
+ SimpleString address = RandomUtil.randomSimpleString();
+ SimpleString queue = RandomUtil.randomSimpleString();
+ server.createQueue(address, routingType, queue, null, true, false);
+ ServerLocator locator = createInVMNonHALocator();
+ ClientSessionFactory cf = createSessionFactory(locator);
+ ClientSession s = addClientSession(cf.createSession(true, false));
+ ClientProducer p = s.createProducer(address);
+ for (int j = 0; j < MESSAGE_COUNT; j++) {
+ p.send(s.createMessage(true).setRoutingType(routingType));
+ }
+ p.close();
+ s.close();
+ cf.close();
+ locator.close();
+ Wait.assertTrue(() -> server.locateQueue(getDefaultExpiryQueueName(address)) != null, 2000, 10);
+ Queue expiry = server.locateQueue(AddressSettings.DEFAULT_EXPIRY_QUEUE_PREFIX.concat(address).concat(AddressSettings.DEFAULT_EXPIRY_QUEUE_SUFFIX));
+ Wait.assertEquals(MESSAGE_COUNT, expiry::getMessageCount);
+ }
+
+ assertEquals(ITERATIONS, server.getPostOffice().getBindingsForAddress(expiryAddress).getBindings().size());
+ }
+
+ @Test
+ public void testAutoDeletionAndRecreationOfExpiryResources() throws Exception {
+ SimpleString expiryQueueName = getDefaultExpiryQueueName(addressA);
+
+ triggerExpiration();
+
+ // consume the message from the DLQ so it will be auto-deleted
+ ServerLocator locator = createInVMNonHALocator();
+ ClientSessionFactory sessionFactory = createSessionFactory(locator);
+ ClientSession session = addClientSession(sessionFactory.createSession(true, true));
+ Wait.assertTrue(() -> server.locateQueue(expiryQueueName) != null, 2000, 100);
+ ClientConsumer consumer = session.createConsumer(expiryQueueName);
+ session.start();
+ ClientMessage message = consumer.receive();
+ assertNotNull(message);
+ message.acknowledge();
+ consumer.close();
+ session.close();
+ sessionFactory.close();
+ locator.close();
+
+ Wait.assertTrue(() -> server.locateQueue(expiryQueueName) == null, 2000, 100);
+
+ server.destroyQueue(queueA);
+
+ triggerExpiration();
+ Wait.assertTrue(() -> server.getAddressInfo(expiryAddress) != null, 2000, 100);
+ Wait.assertTrue(() -> server.locateQueue(expiryQueueName) != null, 2000, 100);
+ }
+
+ @Test
+ public void testWithJMSFQQN() throws Exception {
+ SimpleString expiryQueueName = getDefaultExpiryQueueName(addressA);
+ String fqqn = CompositeAddress.toFullyQualified(expiryAddress, expiryQueueName).toString();
+
+ triggerExpiration();
+
+ JMSContext context = new ActiveMQConnectionFactory("vm://0").createContext();
+ context.start();
+ assertNotNull(context.createConsumer(context.createQueue(fqqn)).receive(2000));
+ }
+
+ private SimpleString getDefaultExpiryQueueName(SimpleString address) {
+ return AddressSettings.DEFAULT_EXPIRY_QUEUE_PREFIX.concat(address).concat(AddressSettings.DEFAULT_EXPIRY_QUEUE_SUFFIX);
+ }
+
+ private void triggerExpiration() throws Exception {
+ server.createQueue(addressA, RoutingType.ANYCAST, queueA, null, true, false);
+ ServerLocator locator = createInVMNonHALocator();
+ ClientSessionFactory sessionFactory = createSessionFactory(locator);
+ ClientSession session = addClientSession(sessionFactory.createSession(true, false));
+ ClientProducer producer = addClientProducer(session.createProducer(addressA));
+ ClientMessage message = session.createMessage(true);
+ producer.send(message);
+ producer.close();
+ session.close();
+ sessionFactory.close();
+ locator.close();
+ }
+}