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/02/27 14:41:45 UTC
[activemq-artemis] branch master updated: ARTEMIS-2263 Support
queue level auto-delete configuration
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
The following commit(s) were added to refs/heads/master by this push:
new 9b01e95 ARTEMIS-2263 Support queue level auto-delete configuration
new ea7a9ac This closes #2569
9b01e95 is described below
commit 9b01e9521c89105b4447aba7199b058478986c4f
Author: Michael André Pearce <mi...@me.com>
AuthorDate: Tue Feb 26 17:37:31 2019 +0000
ARTEMIS-2263 Support queue level auto-delete configuration
Add ability to configure when creating auto created queues at the queue level
Add support for configuring message count check
Add test cases
Update docs
---
.../activemq/artemis/api/core/QueueAttributes.java | 40 +++
.../api/config/ActiveMQDefaultConfiguration.java | 18 ++
.../artemis/api/core/client/ClientSession.java | 6 +
.../api/core/management/ActiveMQServerControl.java | 3 +
.../artemis/core/client/impl/QueueQueryImpl.java | 29 +-
.../protocol/core/impl/ActiveMQSessionContext.java | 5 +-
.../impl/wireformat/CreateQueueMessage_V2.java | 72 ++++-
.../wireformat/CreateSharedQueueMessage_V2.java | 61 ++++
.../SessionQueueQueryResponseMessage_V3.java | 62 +++-
.../artemis/core/server/QueueQueryResult.java | 27 ++
.../deployers/impl/FileConfigurationParser.java | 6 +
.../management/impl/ActiveMQServerControlImpl.java | 11 +-
.../artemis/core/persistence/QueueBindingInfo.java | 6 +
.../journal/AbstractJournalStorageManager.java | 2 +-
.../codec/PersistentQueueBindingEncoding.java | 56 +++-
.../core/postoffice/impl/PostOfficeImpl.java | 65 ++---
.../protocol/core/ServerSessionPacketHandler.java | 6 +-
.../artemis/core/server/ActiveMQServer.java | 10 +-
.../apache/activemq/artemis/core/server/Queue.java | 6 +
.../activemq/artemis/core/server/QueueConfig.java | 57 +++-
.../artemis/core/server/ServerSession.java | 8 +-
.../core/server/impl/ActiveMQServerImpl.java | 56 ++--
.../artemis/core/server/impl/LastValueQueue.java | 5 +-
.../core/server/impl/PostOfficeJournalLoader.java | 3 +
.../artemis/core/server/impl/QueueFactoryImpl.java | 6 +-
.../artemis/core/server/impl/QueueImpl.java | 33 ++-
.../artemis/core/server/impl/QueueManagerImpl.java | 64 ++--
.../core/server/impl/ServerSessionImpl.java | 37 ++-
.../core/settings/impl/AddressSettings.java | 35 ++-
.../resources/schema/artemis-configuration.xsd | 9 +-
.../server/impl/ScheduledDeliveryHandlerTest.java | 18 ++
docs/user-manual/en/address-model.md | 22 +-
.../jms/client/ConsumerDelayDispatchTest.java | 4 +-
.../tests/integration/jms/client/GroupingTest.java | 6 +-
.../jms/client/QueueAutoDeleteTest.java | 325 +++++++++++++++++++++
.../ActiveMQServerControlUsingCoreTest.java | 2 +-
.../persistence/QueueConfigRestartTest.java | 4 +-
.../tests/unit/core/postoffice/impl/FakeQueue.java | 18 ++
38 files changed, 1073 insertions(+), 130 deletions(-)
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/QueueAttributes.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/QueueAttributes.java
index 75b5775..b65a3c3 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/QueueAttributes.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/QueueAttributes.java
@@ -35,6 +35,9 @@ public class QueueAttributes implements Serializable {
public static final String CONSUMERS_BEFORE_DISPATCH = "consumers-before-dispatch";
public static final String DELAY_BEFORE_DISPATCH = "delay-before-dispatch";
public static final String CONSUMER_PRIORITY = "consumer-priority";
+ public static final String AUTO_DELETE = "auto-delete";
+ public static final String AUTO_DELETE_DELAY = "auto-delete-delay";
+ public static final String AUTO_DELETE_MESSAGE_COUNT = "auto-delete-message-count";
private RoutingType routingType;
private SimpleString filterString;
@@ -50,6 +53,10 @@ public class QueueAttributes implements Serializable {
private Integer consumersBeforeDispatch;
private Long delayBeforeDispatch;
private Integer consumerPriority;
+ private Boolean autoDelete;
+ private Long autoDeleteDelay;
+ private Long autoDeleteMessageCount;
+
public void set(String key, String value) {
if (key != null && value != null) {
@@ -81,6 +88,12 @@ public class QueueAttributes implements Serializable {
setGroupRebalance(Boolean.valueOf(value));
} else if (key.equals(GROUP_BUCKETS)) {
setGroupBuckets(Integer.valueOf(value));
+ } else if (key.equals(AUTO_DELETE)) {
+ setAutoDelete(Boolean.valueOf(value));
+ } else if (key.equals(AUTO_DELETE_DELAY)) {
+ setAutoDeleteDelay(Long.valueOf(value));
+ } else if (key.equals(AUTO_DELETE_MESSAGE_COUNT)) {
+ setAutoDeleteMessageCount(Long.valueOf(value));
}
}
}
@@ -210,4 +223,31 @@ public class QueueAttributes implements Serializable {
this.groupBuckets = groupBuckets;
return this;
}
+
+ public Boolean getAutoDelete() {
+ return autoDelete;
+ }
+
+ public QueueAttributes setAutoDelete(Boolean autoDelete) {
+ this.autoDelete = autoDelete;
+ return this;
+ }
+
+ public Long getAutoDeleteDelay() {
+ return autoDeleteDelay;
+ }
+
+ public QueueAttributes setAutoDeleteDelay(Long autoDeleteDelay) {
+ this.autoDeleteDelay = autoDeleteDelay;
+ return this;
+ }
+
+ public Long getAutoDeleteMessageCount() {
+ return autoDeleteMessageCount;
+ }
+
+ public QueueAttributes setAutoDeleteMessageCount(Long autoDeleteMessageCount) {
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
+ return this;
+ }
}
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 d85b978..e0765d0 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
@@ -501,6 +501,12 @@ public final class ActiveMQDefaultConfiguration {
public static final boolean DEFAULT_PURGE_ON_NO_CONSUMERS = false;
+ public static final boolean DEFAULT_QUEUE_AUTO_DELETE = true;
+
+ public static final long DEFAULT_QUEUE_AUTO_DELETE_DELAY = 0;
+
+ public static final long DEFAULT_QUEUE_AUTO_DELETE_MESSAGE_COUNT = 0;
+
public static final int DEFAULT_CONSUMERS_BEFORE_DISPATCH = 0;
public static final long DEFAULT_DELAY_BEFORE_DISPATCH = -1;
@@ -1366,6 +1372,18 @@ public final class ActiveMQDefaultConfiguration {
return DEFAULT_PURGE_ON_NO_CONSUMERS;
}
+ public static boolean getDefaultQueueAutoDelete() {
+ return DEFAULT_QUEUE_AUTO_DELETE;
+ }
+
+ public static long getDefaultQueueAutoDeleteDelay() {
+ return DEFAULT_QUEUE_AUTO_DELETE_DELAY;
+ }
+
+ public static long getDefaultQueueAutoDeleteMessageCount() {
+ return DEFAULT_QUEUE_AUTO_DELETE_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/client/ClientSession.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/ClientSession.java
index 9326e30..70bbf65 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/ClientSession.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/ClientSession.java
@@ -162,6 +162,12 @@ public interface ClientSession extends XAResource, AutoCloseable {
Boolean isGroupRebalance();
Integer getGroupBuckets();
+
+ Boolean isAutoDelete();
+
+ Long getAutoDeleteDelay();
+
+ Long getAutoDeleteMessageCount();
}
// Lifecycle operations ------------------------------------------
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 a6dc829..d93fef2 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
@@ -605,6 +605,9 @@ public interface ActiveMQServerControl {
@Parameter(name = "nonDestructive", desc = "If the queue is non-destructive") boolean nonDestructive,
@Parameter(name = "consumersBeforeDispatch", desc = "Number of consumers needed before dispatch can start") int consumersBeforeDispatch,
@Parameter(name = "delayBeforeDispatch", desc = "Delay to wait before dispatching if number of consumers before dispatch is not met") long delayBeforeDispatch,
+ @Parameter(name = "autoDelete", desc = "If the queue should be deleted once no consumers") boolean autoDelete,
+ @Parameter(name = "autoDeleteDelay", desc = "How long to wait (in milliseconds) before deleting auto-created queues after the queue has 0 consumers") long autoDeleteDelay,
+ @Parameter(name = "autoDeleteMessageCount", desc = "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") long autoDeleteMessageCount,
@Parameter(name = "autoCreateAddress", desc = "Create an address with default values should a matching address not be found") boolean autoCreateAddress) throws Exception;
/**
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/QueueQueryImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/QueueQueryImpl.java
index bef8b37..e017584 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/QueueQueryImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/QueueQueryImpl.java
@@ -64,6 +64,12 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
private final Long delayBeforeDispatch;
+ private final Boolean autoDelete;
+
+ private final Long autoDeleteDelay;
+
+ private final Long autoDeleteMessageCount;
+
private final Integer defaultConsumerWindowSize;
@@ -122,7 +128,7 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
final Boolean exclusive,
final Boolean lastValue,
final Integer defaultConsumerWindowSize) {
- this(durable, temporary, consumerCount, messageCount, filterString, address, name, exists, autoCreateQueues, maxConsumers, autoCreated, purgeOnNoConsumers, routingType, exclusive, null, null, lastValue, null, null, null, null, defaultConsumerWindowSize);
+ this(durable, temporary, consumerCount, messageCount, filterString, address, name, exists, autoCreateQueues, maxConsumers, autoCreated, purgeOnNoConsumers, routingType, exclusive, null, null, lastValue, null, null, null, null, null, null, null, defaultConsumerWindowSize);
}
public QueueQueryImpl(final boolean durable,
@@ -146,6 +152,9 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
final Boolean nonDestructive,
final Integer consumersBeforeDispatch,
final Long delayBeforeDispatch,
+ final Boolean autoDelete,
+ final Long autoDeleteDelay,
+ final Long autoDeleteMessageCount,
final Integer defaultConsumerWindowSize) {
this.durable = durable;
this.temporary = temporary;
@@ -168,6 +177,9 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
this.nonDestructive = nonDestructive;
this.consumersBeforeDispatch = consumersBeforeDispatch;
this.delayBeforeDispatch = delayBeforeDispatch;
+ this.autoDelete = autoDelete;
+ this.autoDeleteDelay = autoDeleteDelay;
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
this.defaultConsumerWindowSize = defaultConsumerWindowSize;
}
@@ -280,5 +292,20 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
public Integer getGroupBuckets() {
return groupBuckets;
}
+
+ @Override
+ public Boolean isAutoDelete() {
+ return autoDelete;
+ }
+
+ @Override
+ public Long getAutoDeleteDelay() {
+ return autoDeleteDelay;
+ }
+
+ @Override
+ public Long getAutoDeleteMessageCount() {
+ return autoDeleteMessageCount;
+ }
}
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
index 71e363e..6f16762 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
@@ -321,6 +321,9 @@ public class ActiveMQSessionContext extends SessionContext {
queueAttributes.getNonDestructive(),
queueAttributes.getConsumersBeforeDispatch(),
queueAttributes.getDelayBeforeDispatch(),
+ queueAttributes.getAutoDelete(),
+ queueAttributes.getAutoDeleteDelay(),
+ queueAttributes.getAutoDeleteMessageCount(),
true), PacketImpl.NULL_RESPONSE);
}
@@ -873,7 +876,7 @@ public class ActiveMQSessionContext extends SessionContext {
// We try to recreate any non-durable or auto-created queues, since they might not be there on failover/reconnect.
// This allows e.g. JMS non durable subs and temporary queues to continue to be used after failover/reconnection
if (!queueInfo.isDurable() || queueInfo.isAutoCreated()) {
- CreateQueueMessage_V2 createQueueRequest = new CreateQueueMessage_V2(queueInfo.getAddress(), queueInfo.getName(), queueInfo.getRoutingType(), queueInfo.getFilterString(), queueInfo.isDurable(), queueInfo.isTemporary(), queueInfo.getMaxConsumers(), queueInfo.isPurgeOnNoConsumers(), queueInfo.isAutoCreated(), false, queueInfo.isExclusive(), queueInfo.isGroupRebalance(), queueInfo.getGroupBuckets(), queueInfo.isLastValue(), queueInfo.getLastValueKey(), queueInfo.isNonDestructive(), [...]
+ CreateQueueMessage_V2 createQueueRequest = new CreateQueueMessage_V2(queueInfo.getAddress(), queueInfo.getName(), queueInfo.getRoutingType(), queueInfo.getFilterString(), queueInfo.isDurable(), queueInfo.isTemporary(), queueInfo.getMaxConsumers(), queueInfo.isPurgeOnNoConsumers(), queueInfo.isAutoCreated(), false, queueInfo.isExclusive(), queueInfo.isGroupRebalance(), queueInfo.getGroupBuckets(), queueInfo.isLastValue(), queueInfo.getLastValueKey(), queueInfo.isNonDestructive(), [...]
sendPacketWithoutLock(sessionChannel, createQueueRequest);
}
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage_V2.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage_V2.java
index 4a383b3..ef8cce2 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage_V2.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage_V2.java
@@ -48,6 +48,12 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
private Long delayBeforeDispatch;
+ private Boolean autoDelete;
+
+ private Long autoDeleteDelay;
+
+ private Long autoDeleteMessageCount;
+
public CreateQueueMessage_V2(final SimpleString address,
final SimpleString queueName,
final boolean temporary,
@@ -72,7 +78,10 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
queueAttributes.getLastValueKey(),
queueAttributes.getNonDestructive(),
queueAttributes.getConsumersBeforeDispatch(),
- queueAttributes.getDelayBeforeDispatch()
+ queueAttributes.getDelayBeforeDispatch(),
+ queueAttributes.getAutoDelete(),
+ queueAttributes.getAutoDeleteDelay(),
+ queueAttributes.getAutoDeleteMessageCount()
);
}
@@ -93,7 +102,10 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
final SimpleString lastValueKey,
final Boolean nonDestructive,
final Integer consumersBeforeDispatch,
- final Long delayBeforeDispatch) {
+ final Long delayBeforeDispatch,
+ final Boolean autoDelete,
+ final Long autoDeleteDelay,
+ final Long autoDeleteMessageCount) {
this();
this.address = address;
@@ -114,6 +126,9 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
this.nonDestructive = nonDestructive;
this.consumersBeforeDispatch = consumersBeforeDispatch;
this.delayBeforeDispatch = delayBeforeDispatch;
+ this.autoDelete = autoDelete;
+ this.autoDeleteDelay = autoDeleteDelay;
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
}
public CreateQueueMessage_V2() {
@@ -137,6 +152,10 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
buff.append(", nonDestructive=" + nonDestructive);
buff.append(", consumersBeforeDispatch=" + consumersBeforeDispatch);
buff.append(", delayBeforeDispatch=" + delayBeforeDispatch);
+ buff.append(", autoDelete=" + autoDelete);
+ buff.append(", autoDeleteDelay=" + autoDeleteDelay);
+ buff.append(", autoDeleteMessageCount=" + autoDeleteMessageCount);
+
buff.append("]");
return buff.toString();
}
@@ -237,6 +256,30 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
this.groupBuckets = groupBuckets;
}
+ public Boolean isAutoDelete() {
+ return autoDelete;
+ }
+
+ public void setAutoDelete(Boolean autoDelete) {
+ this.autoDelete = autoDelete;
+ }
+
+ public Long getAutoDeleteDelay() {
+ return autoDeleteDelay;
+ }
+
+ public void setAutoDeleteDelay(Long autoDeleteDelay) {
+ this.autoDeleteDelay = autoDeleteDelay;
+ }
+
+ public Long getAutoDeleteMessageCount() {
+ return autoDeleteMessageCount;
+ }
+
+ public void setAutoDeleteMessageCount(Long autoDeleteMessageCount) {
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
+ }
+
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
super.encodeRest(buffer);
@@ -252,6 +295,10 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
BufferHelper.writeNullableLong(buffer, delayBeforeDispatch);
BufferHelper.writeNullableBoolean(buffer, groupRebalance);
BufferHelper.writeNullableInteger(buffer, groupBuckets);
+ BufferHelper.writeNullableBoolean(buffer, autoDelete);
+ BufferHelper.writeNullableLong(buffer, autoDeleteDelay);
+ BufferHelper.writeNullableLong(buffer, autoDeleteMessageCount);
+
}
@Override
@@ -272,6 +319,9 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
delayBeforeDispatch = BufferHelper.readNullableLong(buffer);
groupRebalance = BufferHelper.readNullableBoolean(buffer);
groupBuckets = BufferHelper.readNullableInteger(buffer);
+ autoDelete = BufferHelper.readNullableBoolean(buffer);
+ autoDeleteDelay = BufferHelper.readNullableLong(buffer);
+ autoDeleteMessageCount = BufferHelper.readNullableLong(buffer);
}
}
@@ -291,6 +341,9 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
result = prime * result + (nonDestructive == null ? 0 : nonDestructive ? 1231 : 1237);
result = prime * result + (consumersBeforeDispatch == null ? 0 : consumersBeforeDispatch.hashCode());
result = prime * result + (delayBeforeDispatch == null ? 0 : delayBeforeDispatch.hashCode());
+ result = prime * result + (autoDelete == null ? 0 : autoDelete.hashCode());
+ result = prime * result + (autoDeleteDelay == null ? 0 : autoDeleteDelay.hashCode());
+ result = prime * result + (autoDeleteMessageCount == null ? 0 : autoDeleteMessageCount.hashCode());
return result;
}
@@ -349,6 +402,21 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
return false;
} else if (!delayBeforeDispatch.equals(other.delayBeforeDispatch))
return false;
+ if (autoDelete == null) {
+ if (other.autoDelete != null)
+ return false;
+ } else if (!autoDelete.equals(other.autoDelete))
+ return false;
+ if (autoDeleteDelay == null) {
+ if (other.autoDeleteDelay != null)
+ return false;
+ } else if (!autoDeleteDelay.equals(other.autoDeleteDelay))
+ return false;
+ if (autoDeleteMessageCount == null) {
+ if (other.autoDeleteMessageCount != null)
+ return false;
+ } else if (!autoDeleteMessageCount.equals(other.autoDeleteMessageCount))
+ return false;
if (routingType == null) {
if (other.routingType != null)
return false;
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage_V2.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage_V2.java
index dcd936a..f1cff91 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage_V2.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage_V2.java
@@ -34,6 +34,9 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
private Boolean nonDestructive;
private Integer consumersBeforeDispatch;
private Long delayBeforeDispatch;
+ private Boolean autoDelete;
+ private Long autoDeleteDelay;
+ private Long autoDeleteMessageCount;
public CreateSharedQueueMessage_V2(final SimpleString address,
final SimpleString queueName,
@@ -50,6 +53,9 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
final Boolean nonDestructive,
final Integer consumersBeforeDispatch,
final Long delayBeforeDispatch,
+ final Boolean autoDelete,
+ final Long autoDeleteDelay,
+ final Long autoDeleteMessageCount,
final boolean requiresResponse) {
this();
@@ -68,6 +74,9 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
this.nonDestructive = nonDestructive;
this.consumersBeforeDispatch = consumersBeforeDispatch;
this.delayBeforeDispatch = delayBeforeDispatch;
+ this.autoDelete = autoDelete;
+ this.autoDeleteDelay = autoDeleteDelay;
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
this.requiresResponse = requiresResponse;
}
@@ -163,6 +172,30 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
this.groupBuckets = groupBuckets;
}
+ public Boolean isAutoDelete() {
+ return autoDelete;
+ }
+
+ public void setAutoDelete(Boolean autoDelete) {
+ this.autoDelete = autoDelete;
+ }
+
+ public Long getAutoDeleteDelay() {
+ return autoDeleteDelay;
+ }
+
+ public void setAutoDeleteDelay(Long autoDeleteDelay) {
+ this.autoDeleteDelay = autoDeleteDelay;
+ }
+
+ public Long getAutoDeleteMessageCount() {
+ return autoDeleteMessageCount;
+ }
+
+ public void setAutoDeleteMessageCount(Long autoDeleteMessageCount) {
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
+ }
+
@Override
public String toString() {
StringBuffer buff = new StringBuffer(getParentString());
@@ -181,6 +214,9 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
buff.append(", nonDestructive=" + nonDestructive);
buff.append(", consumersBeforeDispatch=" + consumersBeforeDispatch);
buff.append(", delayBeforeDispatch=" + delayBeforeDispatch);
+ buff.append(", autoDelete=" + autoDelete);
+ buff.append(", autoDeleteDelay=" + autoDeleteDelay);
+ buff.append(", autoDeleteMessageCount=" + autoDeleteMessageCount);
buff.append(", requiresResponse=" + requiresResponse);
buff.append("]");
return buff.toString();
@@ -204,6 +240,10 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
BufferHelper.writeNullableLong(buffer, delayBeforeDispatch);
BufferHelper.writeNullableBoolean(buffer, groupRebalance);
BufferHelper.writeNullableInteger(buffer, groupBuckets);
+ BufferHelper.writeNullableBoolean(buffer, autoDelete);
+ BufferHelper.writeNullableLong(buffer, autoDeleteDelay);
+ BufferHelper.writeNullableLong(buffer, autoDeleteMessageCount);
+
}
@Override
@@ -227,6 +267,9 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
delayBeforeDispatch = BufferHelper.readNullableLong(buffer);
groupRebalance = BufferHelper.readNullableBoolean(buffer);
groupBuckets = BufferHelper.readNullableInteger(buffer);
+ autoDelete = BufferHelper.readNullableBoolean(buffer);
+ autoDeleteDelay = BufferHelper.readNullableLong(buffer);
+ autoDeleteMessageCount = BufferHelper.readNullableLong(buffer);
}
}
@@ -250,6 +293,9 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
result = prime * result + (nonDestructive == null ? 0 : nonDestructive ? 1231 : 1237);
result = prime * result + (consumersBeforeDispatch == null ? 0 : consumersBeforeDispatch.hashCode());
result = prime * result + (delayBeforeDispatch == null ? 0 : delayBeforeDispatch.hashCode());
+ result = prime * result + (autoDelete == null ? 0 : autoDelete.hashCode());
+ result = prime * result + (autoDeleteDelay == null ? 0 : autoDeleteDelay.hashCode());
+ result = prime * result + (autoDeleteMessageCount == null ? 0 : autoDeleteMessageCount.hashCode());
return result;
}
@@ -334,6 +380,21 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
return false;
} else if (!delayBeforeDispatch.equals(other.delayBeforeDispatch))
return false;
+ if (autoDelete == null) {
+ if (other.autoDelete != null)
+ return false;
+ } else if (!autoDelete.equals(other.autoDelete))
+ return false;
+ if (autoDeleteDelay == null) {
+ if (other.autoDeleteDelay != null)
+ return false;
+ } else if (!autoDeleteDelay.equals(other.autoDeleteDelay))
+ return false;
+ if (autoDeleteMessageCount == null) {
+ if (other.autoDeleteMessageCount != null)
+ return false;
+ } else if (!autoDeleteMessageCount.equals(other.autoDeleteMessageCount))
+ return false;
return true;
}
}
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionQueueQueryResponseMessage_V3.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionQueueQueryResponseMessage_V3.java
index 09bdaca..1a3a903 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionQueueQueryResponseMessage_V3.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionQueueQueryResponseMessage_V3.java
@@ -50,14 +50,20 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
private Long delayBeforeDispatch;
+ private Boolean autoDelete;
+
+ private Long autoDeleteDelay;
+
+ private Long autoDeleteMessageCount;
+
protected Integer defaultConsumerWindowSize;
public SessionQueueQueryResponseMessage_V3(final QueueQueryResult result) {
- this(result.getName(), result.getAddress(), result.isDurable(), result.isTemporary(), result.getFilterString(), result.getConsumerCount(), result.getMessageCount(), result.isExists(), result.isAutoCreateQueues(), result.isAutoCreated(), result.isPurgeOnNoConsumers(), result.getRoutingType(), result.getMaxConsumers(), result.isExclusive(), result.isGroupRebalance(), result.getGroupBuckets(), result.isLastValue(), result.getLastValueKey(), result.isNonDestructive(), result.getConsume [...]
+ this(result.getName(), result.getAddress(), result.isDurable(), result.isTemporary(), result.getFilterString(), result.getConsumerCount(), result.getMessageCount(), result.isExists(), result.isAutoCreateQueues(), result.isAutoCreated(), result.isPurgeOnNoConsumers(), result.getRoutingType(), result.getMaxConsumers(), result.isExclusive(), result.isGroupRebalance(), result.getGroupBuckets(), result.isLastValue(), result.getLastValueKey(), result.isNonDestructive(), result.getConsume [...]
}
public SessionQueueQueryResponseMessage_V3() {
- this(null, null, false, false, null, 0, 0, false, false, false, false, RoutingType.MULTICAST, -1, null, null,null, null, null, null, null, null, null);
+ this(null, null, false, false, null, 0, 0, false, false, false, false, RoutingType.MULTICAST, -1, null, null,null, null, null, null, null, null, null, null, null, null);
}
private SessionQueueQueryResponseMessage_V3(final SimpleString name,
@@ -81,6 +87,9 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
final Boolean nonDestructive,
final Integer consumersBeforeDispatch,
final Long delayBeforeDispatch,
+ final Boolean autoDelete,
+ final Long autoDeleteDelay,
+ final Long autoDeleteMessageCount,
final Integer defaultConsumerWindowSize) {
super(SESS_QUEUEQUERY_RESP_V3);
@@ -126,6 +135,12 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
this.delayBeforeDispatch = delayBeforeDispatch;
+ this.autoDelete = autoDelete;
+
+ this.autoDeleteDelay = autoDeleteDelay;
+
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
+
this.defaultConsumerWindowSize = defaultConsumerWindowSize;
}
@@ -233,6 +248,18 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
this.groupBuckets = groupBuckets;
}
+ public Boolean isAutoDelete() {
+ return autoDelete;
+ }
+
+ public Long getAutoDeleteDelay() {
+ return autoDeleteDelay;
+ }
+
+ public Long getAutoDeleteMessageCount() {
+ return autoDeleteMessageCount;
+ }
+
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
super.encodeRest(buffer);
@@ -249,6 +276,10 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
BufferHelper.writeNullableLong(buffer, delayBeforeDispatch);
BufferHelper.writeNullableBoolean(buffer, groupRebalance);
BufferHelper.writeNullableInteger(buffer, groupBuckets);
+ BufferHelper.writeNullableBoolean(buffer, autoDelete);
+ BufferHelper.writeNullableLong(buffer, autoDeleteDelay);
+ BufferHelper.writeNullableLong(buffer, autoDeleteMessageCount);
+
}
@Override
@@ -272,6 +303,10 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
delayBeforeDispatch = BufferHelper.readNullableLong(buffer);
groupRebalance = BufferHelper.readNullableBoolean(buffer);
groupBuckets = BufferHelper.readNullableInteger(buffer);
+ autoDelete = BufferHelper.readNullableBoolean(buffer);
+ autoDeleteDelay = BufferHelper.readNullableLong(buffer);
+ autoDeleteMessageCount = BufferHelper.readNullableLong(buffer);
+
}
}
@@ -291,6 +326,9 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
result = prime * result + (nonDestructive == null ? 0 : nonDestructive ? 1231 : 1237);
result = prime * result + (consumersBeforeDispatch == null ? 0 : consumersBeforeDispatch.hashCode());
result = prime * result + (delayBeforeDispatch == null ? 0 : delayBeforeDispatch.hashCode());
+ result = prime * result + (autoDelete == null ? 0 : autoDelete.hashCode());
+ result = prime * result + (autoDeleteDelay == null ? 0 : autoDeleteDelay.hashCode());
+ result = prime * result + (autoDeleteMessageCount == null ? 0 : autoDeleteMessageCount.hashCode());
result = prime * result + ((defaultConsumerWindowSize == null) ? 0 : defaultConsumerWindowSize.hashCode());
return result;
}
@@ -317,13 +355,16 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
buff.append(", nonDestructive=" + nonDestructive);
buff.append(", consumersBeforeDispatch=" + consumersBeforeDispatch);
buff.append(", delayBeforeDispatch=" + delayBeforeDispatch);
+ buff.append(", autoDelete=" + autoDelete);
+ buff.append(", autoDeleteDelay=" + autoDeleteDelay);
+ buff.append(", autoDeleteMessageCount=" + autoDeleteMessageCount);
buff.append(", defaultConsumerWindowSize=" + defaultConsumerWindowSize);
return buff.toString();
}
@Override
public ClientSession.QueueQuery toQueueQuery() {
- return new QueueQueryImpl(isDurable(), isTemporary(), getConsumerCount(), getMessageCount(), getFilterString(), getAddress(), getName(), isExists(), isAutoCreateQueues(), getMaxConsumers(), isAutoCreated(), isPurgeOnNoConsumers(), getRoutingType(), isExclusive(), isGroupRebalance(), getGroupBuckets(), isLastValue(), getLastValueKey(), isNonDestructive(), getConsumersBeforeDispatch(), getDelayBeforeDispatch(), getDefaultConsumerWindowSize());
+ return new QueueQueryImpl(isDurable(), isTemporary(), getConsumerCount(), getMessageCount(), getFilterString(), getAddress(), getName(), isExists(), isAutoCreateQueues(), getMaxConsumers(), isAutoCreated(), isPurgeOnNoConsumers(), getRoutingType(), isExclusive(), isGroupRebalance(), getGroupBuckets(), isLastValue(), getLastValueKey(), isNonDestructive(), getConsumersBeforeDispatch(), getDelayBeforeDispatch(), isAutoDelete(), getAutoDeleteDelay(), getAutoDeleteMessageCount(), getDef [...]
}
@Override
@@ -379,6 +420,21 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
return false;
} else if (!delayBeforeDispatch.equals(other.delayBeforeDispatch))
return false;
+ if (autoDelete == null) {
+ if (other.autoDelete != null)
+ return false;
+ } else if (!autoDelete.equals(other.autoDelete))
+ return false;
+ if (autoDeleteDelay == null) {
+ if (other.autoDeleteDelay != null)
+ return false;
+ } else if (!autoDeleteDelay.equals(other.autoDeleteDelay))
+ return false;
+ if (autoDeleteMessageCount == null) {
+ if (other.autoDeleteMessageCount != null)
+ return false;
+ } else if (!autoDeleteMessageCount.equals(other.autoDeleteMessageCount))
+ return false;
if (defaultConsumerWindowSize == null) {
if (other.defaultConsumerWindowSize != null)
return false;
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/server/QueueQueryResult.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/server/QueueQueryResult.java
index 92bf200..1d84f3e 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/server/QueueQueryResult.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/server/QueueQueryResult.java
@@ -63,6 +63,12 @@ public class QueueQueryResult {
private Long delayBeforeDispatch;
+ private Boolean autoDelete;
+
+ private Long autoDeleteDelay;
+
+ private Long autoDeleteMessageCount;
+
private Integer defaultConsumerWindowSize;
public QueueQueryResult(final SimpleString name,
@@ -86,6 +92,9 @@ public class QueueQueryResult {
final Boolean nonDestructive,
final Integer consumersBeforeDispatch,
final Long delayBeforeDispatch,
+ final Boolean autoDelete,
+ final Long autoDeleteDelay,
+ final Long autoDeleteMessageCount,
final Integer defaultConsumerWindowSize) {
this.durable = durable;
@@ -129,6 +138,12 @@ public class QueueQueryResult {
this.delayBeforeDispatch = delayBeforeDispatch;
+ this.autoDelete = autoDelete;
+
+ this.autoDeleteDelay = autoDeleteDelay;
+
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
+
this.defaultConsumerWindowSize = defaultConsumerWindowSize;
}
@@ -223,4 +238,16 @@ public class QueueQueryResult {
public Integer getGroupBuckets() {
return groupBuckets;
}
+
+ public Boolean isAutoDelete() {
+ return autoDelete;
+ }
+
+ public Long getAutoDeleteDelay() {
+ return autoDeleteDelay;
+ }
+
+ public Long getAutoDeleteMessageCount() {
+ return autoDeleteMessageCount;
+ }
}
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 e5aba7a..332938d 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
@@ -219,6 +219,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
private static final String AUTO_DELETE_QUEUES_DELAY = "auto-delete-queues-delay";
+ private static final String AUTO_DELETE_QUEUES_MESSAGE_COUNT = "auto-delete-queues-message-count";
+
private static final String CONFIG_DELETE_QUEUES = "config-delete-queues";
private static final String AUTO_CREATE_ADDRESSES = "auto-create-addresses";
@@ -1061,6 +1063,10 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
long autoDeleteQueuesDelay = XMLUtil.parseLong(child);
Validators.GE_ZERO.validate(AUTO_DELETE_QUEUES_DELAY, autoDeleteQueuesDelay);
addressSettings.setAutoDeleteQueuesDelay(autoDeleteQueuesDelay);
+ } else if (AUTO_DELETE_QUEUES_MESSAGE_COUNT.equalsIgnoreCase(name)) {
+ long autoDeleteQueuesMessageCount = XMLUtil.parseLong(child);
+ Validators.MINUS_ONE_OR_GE_ZERO.validate(AUTO_DELETE_QUEUES_MESSAGE_COUNT, autoDeleteQueuesMessageCount);
+ addressSettings.setAutoDeleteQueuesMessageCount(autoDeleteQueuesMessageCount);
} else if (CONFIG_DELETE_QUEUES.equalsIgnoreCase(name)) {
String value = getTrimmedTextContent(child);
Validators.DELETION_POLICY_TYPE.validate(CONFIG_DELETE_QUEUES, value);
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 f71711e..318d880 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
@@ -833,7 +833,11 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
addressSettings.getDefaultLastValueKey() == null ? null : addressSettings.getDefaultLastValueKey().toString(),
addressSettings.isDefaultNonDestructive(),
addressSettings.getDefaultConsumersBeforeDispatch(),
- addressSettings.getDefaultDelayBeforeDispatch(), autoCreateAddress
+ addressSettings.getDefaultDelayBeforeDispatch(),
+ addressSettings.isAutoDeleteQueues(),
+ addressSettings.getAutoDeleteQueuesDelay(),
+ addressSettings.getAutoDeleteQueuesMessageCount(),
+ autoCreateAddress
);
}
@@ -853,6 +857,9 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
boolean nonDestructive,
int consumersBeforeDispatch,
long delayBeforeDispatch,
+ boolean autoDelete,
+ long autoDeleteDelay,
+ long autoDeleteMessageCount,
boolean autoCreateAddress) throws Exception {
checkStarted();
@@ -864,7 +871,7 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
filter = new SimpleString(filterStr);
}
- final Queue queue = server.createQueue(SimpleString.toSimpleString(address), RoutingType.valueOf(routingType.toUpperCase()), SimpleString.toSimpleString(name), filter, durable, false, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, SimpleString.toSimpleString(lastValueKey), nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoCreateAddress);
+ final Queue queue = server.createQueue(SimpleString.toSimpleString(address), RoutingType.valueOf(routingType.toUpperCase()), SimpleString.toSimpleString(name), filter, durable, false, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, SimpleString.toSimpleString(lastValueKey), nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoDelete, autoDeleteDelay, autoDeleteMessageCount, autoCreateAddress);
return QueueTextFormatter.Long.format(queue, new StringBuilder()).toString();
} catch (ActiveMQException e) {
throw new IllegalStateException(e.getMessage());
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java
index 89034d6..ec2823e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java
@@ -89,4 +89,10 @@ public interface QueueBindingInfo {
boolean isGroupRebalance();
int getGroupBuckets();
+
+ boolean isAutoDelete();
+
+ long getAutoDeleteDelay();
+
+ long getAutoDeleteMessageCount();
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
index b441e3a..cbe6512 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
@@ -1297,7 +1297,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
SimpleString filterString = filter == null ? null : filter.getFilterString();
- PersistentQueueBindingEncoding bindingEncoding = new PersistentQueueBindingEncoding(queue.getName(), binding.getAddress(), filterString, queue.getUser(), queue.isAutoCreated(), queue.getMaxConsumers(), queue.isPurgeOnNoConsumers(), queue.isExclusive(), queue.isGroupRebalance(), queue.getGroupBuckets(), queue.isLastValue(), queue.getLastValueKey(), queue.isNonDestructive(), queue.getConsumersBeforeDispatch(), queue.getDelayBeforeDispatch(), queue.getRoutingType().getType(), queue.is [...]
+ PersistentQueueBindingEncoding bindingEncoding = new PersistentQueueBindingEncoding(queue.getName(), binding.getAddress(), filterString, queue.getUser(), queue.isAutoCreated(), queue.getMaxConsumers(), queue.isPurgeOnNoConsumers(), queue.isExclusive(), queue.isGroupRebalance(), queue.getGroupBuckets(), queue.isLastValue(), queue.getLastValueKey(), queue.isNonDestructive(), queue.getConsumersBeforeDispatch(), queue.getDelayBeforeDispatch(), queue.isAutoDelete(), queue.getAutoDeleteD [...]
readLock();
try {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java
index 08236ab..b72ad7f 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java
@@ -66,6 +66,12 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin
public int groupBuckets;
+ public boolean autoDelete;
+
+ public long autoDeleteDelay;
+
+ public long autoDeleteMessageCount;
+
public PersistentQueueBindingEncoding() {
}
@@ -106,6 +112,12 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin
groupRebalance +
", groupBuckets=" +
groupBuckets +
+ ", autoDelete=" +
+ autoDelete +
+ ", autoDeleteDelay=" +
+ autoDeleteDelay +
+ ", autoDeleteMessageCount=" +
+ autoDeleteMessageCount +
"]";
}
@@ -124,6 +136,9 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin
final boolean nonDestructive,
final int consumersBeforeDispatch,
final long delayBeforeDispatch,
+ final boolean autoDelete,
+ final long autoDeleteDelay,
+ final long autoDeleteMessageCount,
final byte routingType,
final boolean configurationManaged) {
this.name = name;
@@ -141,6 +156,9 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin
this.nonDestructive = nonDestructive;
this.consumersBeforeDispatch = consumersBeforeDispatch;
this.delayBeforeDispatch = delayBeforeDispatch;
+ this.autoDelete = autoDelete;
+ this.autoDeleteDelay = autoDeleteDelay;
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
this.routingType = routingType;
this.configurationManaged = configurationManaged;
}
@@ -308,6 +326,21 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin
}
@Override
+ public boolean isAutoDelete() {
+ return autoDelete;
+ }
+
+ @Override
+ public long getAutoDeleteDelay() {
+ return autoDeleteDelay;
+ }
+
+ @Override
+ public long getAutoDeleteMessageCount() {
+ return autoDeleteMessageCount;
+ }
+
+ @Override
public void decode(final ActiveMQBuffer buffer) {
name = buffer.readSimpleString();
address = buffer.readSimpleString();
@@ -383,6 +416,21 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin
} else {
groupBuckets = ActiveMQDefaultConfiguration.getDefaultGroupBuckets();
}
+ if (buffer.readableBytes() > 0) {
+ autoDelete = buffer.readBoolean();
+ } else {
+ autoDelete = ActiveMQDefaultConfiguration.getDefaultQueueAutoDelete();
+ }
+ if (buffer.readableBytes() > 0) {
+ autoDeleteDelay = buffer.readLong();
+ } else {
+ autoDeleteDelay = ActiveMQDefaultConfiguration.getDefaultQueueAutoDeleteDelay();
+ }
+ if (buffer.readableBytes() > 0) {
+ autoDeleteMessageCount = buffer.readLong();
+ } else {
+ autoDeleteMessageCount = ActiveMQDefaultConfiguration.getDefaultQueueAutoDeleteMessageCount();
+ }
}
@Override
@@ -404,6 +452,9 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin
buffer.writeBoolean(nonDestructive);
buffer.writeBoolean(groupRebalance);
buffer.writeInt(groupBuckets);
+ buffer.writeBoolean(autoDelete);
+ buffer.writeLong(autoDeleteDelay);
+ buffer.writeLong(autoDeleteMessageCount);
}
@Override
@@ -422,7 +473,10 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin
SimpleString.sizeofNullableString(lastValueKey) +
DataConstants.SIZE_BOOLEAN +
DataConstants.SIZE_BOOLEAN +
- DataConstants.SIZE_INT;
+ DataConstants.SIZE_INT +
+ DataConstants.SIZE_BOOLEAN +
+ DataConstants.SIZE_LONG +
+ DataConstants.SIZE_LONG;
}
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 d209cf7..ec7675e 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
@@ -72,6 +72,7 @@ import org.apache.activemq.artemis.core.server.RoutingContext;
import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
import org.apache.activemq.artemis.core.server.group.GroupingHandler;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.server.impl.QueueManagerImpl;
import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl;
import org.apache.activemq.artemis.core.server.management.ManagementService;
import org.apache.activemq.artemis.core.server.management.Notification;
@@ -1557,19 +1558,7 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
public void run() {
// The reaper thread should be finished case the PostOffice is gone
// This is to avoid leaks on PostOffice between stops and starts
- Map<SimpleString, Binding> nameMap = addressManager.getBindings();
-
- List<Queue> queues = new ArrayList<>();
-
- for (Binding binding : nameMap.values()) {
- if (binding.getType() == BindingType.LOCAL_QUEUE) {
- Queue queue = (Queue) binding.getBindable();
-
- queues.add(queue);
- }
- }
-
- for (Queue queue : queues) {
+ for (Queue queue : getLocalQueues()) {
try {
queue.expireReferences();
} catch (Exception e) {
@@ -1591,38 +1580,9 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
@Override
public void run() {
- Map<SimpleString, Binding> nameMap = addressManager.getBindings();
-
- List<Queue> queues = new ArrayList<>();
-
- for (Binding binding : nameMap.values()) {
- if (binding.getType() == BindingType.LOCAL_QUEUE) {
- Queue queue = (Queue) binding.getBindable();
-
- queues.add(queue);
- }
- }
-
- for (Queue queue : queues) {
- int consumerCount = queue.getConsumerCount();
- long messageCount = queue.getMessageCount();
- boolean autoCreated = queue.isAutoCreated();
- long consumerRemovedTimestamp = queue.getConsumerRemovedTimestamp();
-
- if (!queue.isInternalQueue() && autoCreated && messageCount == 0 && consumerCount == 0 && consumerRemovedTimestamp != -1) {
- SimpleString queueName = queue.getName();
- AddressSettings settings = addressSettingsRepository.getMatch(queue.getAddress().toString());
- if (settings.isAutoDeleteQueues() && (System.currentTimeMillis() - consumerRemovedTimestamp >= settings.getAutoDeleteQueuesDelay())) {
- if (ActiveMQServerLogger.LOGGER.isDebugEnabled()) {
- ActiveMQServerLogger.LOGGER.info("deleting auto-created queue \"" + queueName + ".\" consumerCount = " + consumerCount + "; messageCount = " + messageCount + "; isAutoDeleteQueues = " + settings.isAutoDeleteQueues());
- }
-
- try {
- server.destroyQueue(queueName, null, true, false, settings.isAutoDeleteAddresses(), true);
- } catch (Exception e) {
- ActiveMQServerLogger.LOGGER.errorRemovingAutoCreatedQueue(e, queueName);
- }
- }
+ for (Queue queue : getLocalQueues()) {
+ if (!queue.isInternalQueue() && QueueManagerImpl.isAutoDelete(queue) && QueueManagerImpl.consumerCountCheck(queue) && QueueManagerImpl.delayCheck(queue) && QueueManagerImpl.messageCountCheck(queue)) {
+ QueueManagerImpl.deleteAutoCreatedQueue(server, queue);
}
}
@@ -1648,6 +1608,21 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
}
}
+ private List<Queue> getLocalQueues() {
+ Map<SimpleString, Binding> nameMap = addressManager.getBindings();
+
+ List<Queue> queues = new ArrayList<>();
+
+ for (Binding binding : nameMap.values()) {
+ if (binding.getType() == BindingType.LOCAL_QUEUE) {
+ Queue queue = (Queue) binding.getBindable();
+
+ queues.add(queue);
+ }
+ }
+ return queues;
+ }
+
public static final class AddOperation implements TransactionOperation {
private final List<MessageReference> refs;
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
index 2aa3046..b1bd4b2 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
@@ -357,7 +357,8 @@ public class ServerSessionPacketHandler implements ChannelHandler {
CreateQueueMessage_V2 request = (CreateQueueMessage_V2) packet;
requiresResponse = request.isRequiresResponse();
session.createQueue(request.getAddress(), request.getQueueName(), request.getRoutingType(), request.getFilterString(), request.isTemporary(), request.isDurable(), request.getMaxConsumers(), request.isPurgeOnNoConsumers(),
- request.isExclusive(), request.isGroupRebalance(), request.getGroupBuckets(), request.isLastValue(), request.getLastValueKey(), request.isNonDestructive(), request.getConsumersBeforeDispatch(), request.getDelayBeforeDispatch(), request.isAutoCreated());
+ request.isExclusive(), request.isGroupRebalance(), request.getGroupBuckets(), request.isLastValue(), request.getLastValueKey(), request.isNonDestructive(), request.getConsumersBeforeDispatch(), request.getDelayBeforeDispatch(),
+ request.isAutoDelete(), request.getAutoDeleteDelay(), request.getAutoDeleteMessageCount(), request.isAutoCreated());
if (requiresResponse) {
response = createNullResponseMessage(packet);
}
@@ -381,7 +382,8 @@ public class ServerSessionPacketHandler implements ChannelHandler {
QueueQueryResult result = session.executeQueueQuery(request.getQueueName());
if (!(result.isExists() && Objects.equals(result.getAddress(), request.getAddress()) && Objects.equals(result.getFilterString(), request.getFilterString()))) {
session.createSharedQueue(request.getAddress(), request.getQueueName(), request.getRoutingType(), request.getFilterString(), request.isDurable(), request.getMaxConsumers(), request.isPurgeOnNoConsumers(),
- request.isExclusive(), request.isGroupRebalance(), request.getGroupBuckets(), request.isLastValue(), request.getLastValueKey(), request.isNonDestructive(), request.getConsumersBeforeDispatch(), request.getDelayBeforeDispatch());
+ request.isExclusive(), request.isGroupRebalance(), request.getGroupBuckets(), request.isLastValue(), request.getLastValueKey(), request.isNonDestructive(), request.getConsumersBeforeDispatch(), request.getDelayBeforeDispatch(),
+ request.isAutoDelete(), request.getAutoDeleteDelay(), request.getAutoDeleteMessageCount());
}
if (requiresResponse) {
response = createNullResponseMessage(packet);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
index eca14b8..b1db8aa 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
@@ -398,7 +398,8 @@ public interface ActiveMQServer extends ServiceComponent {
void createSharedQueue(SimpleString address, RoutingType routingType, SimpleString name, SimpleString filterString,
SimpleString user, boolean durable, int maxConsumers, boolean purgeOnNoConsumers, boolean exclusive,
boolean groupRebalance, int groupBuckets, boolean lastValue,
- SimpleString lastValueKey, boolean nonDestructive, int consumersBeforeDispatch, long delayBeforeDispatch) throws Exception;
+ SimpleString lastValueKey, boolean nonDestructive, int consumersBeforeDispatch, long delayBeforeDispatch,
+ boolean autoDelete, long autoDeleteTimeout, long autoDeleteMessageCount) throws Exception;
Queue createQueue(SimpleString address, RoutingType routingType, SimpleString queueName, SimpleString filter,
boolean durable, boolean temporary) throws Exception;
@@ -412,7 +413,8 @@ public interface ActiveMQServer extends ServiceComponent {
Queue createQueue(SimpleString address, RoutingType routingType, SimpleString queueName, SimpleString filter,
boolean durable, boolean temporary, int maxConsumers, boolean purgeOnNoConsumers, boolean exclusive, boolean groupRebalance, int groupBuckets,
- boolean lastValue, SimpleString lastValueKey, boolean nonDestructive, int consumersBeforeDispatch, long delayBeforeDispatch, boolean autoCreateAddress) throws Exception;
+ boolean lastValue, SimpleString lastValueKey, boolean nonDestructive, int consumersBeforeDispatch, long delayBeforeDispatch,
+ boolean autoDelete, long autoDeleteDelay, long autoDeleteMessageCount, boolean autoCreateAddress) throws Exception;
Queue createQueue(SimpleString address, RoutingType routingType, SimpleString queueName, SimpleString filter,
SimpleString user, boolean durable, boolean temporary, boolean autoCreated, Integer maxConsumers,
@@ -429,7 +431,7 @@ public interface ActiveMQServer extends ServiceComponent {
Queue createQueue(AddressInfo addressInfo, SimpleString queueName, SimpleString filter,
SimpleString user, boolean durable, boolean temporary, boolean autoCreated, Integer maxConsumers,
Boolean purgeOnNoConsumers, Boolean exclusive, Boolean groupRebalance, Integer groupBuckets, Boolean lastValue, SimpleString lastValueKey, Boolean nonDestructive,
- Integer consumersBeforeDispatch, Long delayBeforeDispatch, boolean autoCreateAddress) throws Exception;
+ Integer consumersBeforeDispatch, Long delayBeforeDispatch, Boolean autoDelete, Long autoDeleteDelay, Long autoDeleteMessageCount, boolean autoCreateAddress) throws Exception;
Queue createQueue(SimpleString address, RoutingType routingType, SimpleString queueName, SimpleString filter,
SimpleString user, boolean durable, boolean temporary, boolean ignoreIfExists, boolean transientQueue,
@@ -443,7 +445,7 @@ public interface ActiveMQServer extends ServiceComponent {
SimpleString user, boolean durable, boolean temporary, boolean ignoreIfExists, boolean transientQueue,
boolean autoCreated, int maxConsumers, boolean purgeOnNoConsumers, boolean exclusive, boolean groupRebalance,
int groupBuckets, boolean lastValue, SimpleString lastValueKey, boolean nonDestructive,
- int consumersBeforeDispatch, long delayBeforeDispatch, boolean autoCreateAddress) throws Exception;
+ int consumersBeforeDispatch, long delayBeforeDispatch, boolean autoDelete, long autoDeleteDelay, long autoDeleteMessageCount, boolean autoCreateAddress) throws Exception;
@Deprecated
Queue createQueue(SimpleString address, SimpleString queueName, SimpleString filter, boolean durable, boolean temporary) throws Exception;
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
index 2de830e..2fcd766 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
@@ -66,6 +66,12 @@ public interface Queue extends Bindable,CriticalComponent {
*/
boolean isDurableMessage();
+ boolean isAutoDelete();
+
+ long getAutoDeleteDelay();
+
+ long getAutoDeleteMessageCount();
+
boolean isTemporary();
boolean isAutoCreated();
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/QueueConfig.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/QueueConfig.java
index b3c8009..ebf332d 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/QueueConfig.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/QueueConfig.java
@@ -48,6 +48,9 @@ public final class QueueConfig {
private final boolean configurationManaged;
private final SimpleString lastValueKey;
private final boolean nonDestructive;
+ private final boolean autoDelete;
+ private final long autoDeleteDelay;
+ private final long autoDeleteMessageCount;
public static final class Builder {
@@ -71,6 +74,9 @@ public final class QueueConfig {
private long delayBeforeDispatch;
private boolean groupRebalance;
private int groupBuckets;
+ private boolean autoDelete;
+ private long autoDeleteDelay;
+ private long autoDeleteMessageCount;
private boolean configurationManaged;
private Builder(final long id, final SimpleString name) {
@@ -98,6 +104,9 @@ public final class QueueConfig {
this.delayBeforeDispatch = ActiveMQDefaultConfiguration.getDefaultDelayBeforeDispatch();
this.groupRebalance = ActiveMQDefaultConfiguration.getDefaultGroupRebalance();
this.groupBuckets = ActiveMQDefaultConfiguration.getDefaultGroupBuckets();
+ this.autoDelete = ActiveMQDefaultConfiguration.getDefaultQueueAutoDelete();
+ this.autoDeleteDelay = ActiveMQDefaultConfiguration.getDefaultQueueAutoDeleteDelay();
+ this.autoDeleteMessageCount = ActiveMQDefaultConfiguration.getDefaultQueueAutoDeleteMessageCount();
this.configurationManaged = false;
validateState();
}
@@ -190,6 +199,22 @@ public final class QueueConfig {
return this;
}
+ public Builder autoDelete(final boolean autoDelete) {
+ this.autoDelete = autoDelete;
+ return this;
+ }
+
+ public Builder autoDeleteDelay(final long autoDeleteDelay) {
+ this.autoDeleteDelay = autoDeleteDelay;
+ return this;
+ }
+
+ public Builder autoDeleteMessageCount(final long autoDeleteMessageCount) {
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
+ return this;
+ }
+
+
public Builder groupRebalance(final boolean groupRebalance) {
this.groupRebalance = groupRebalance;
return this;
@@ -233,7 +258,7 @@ public final class QueueConfig {
} else {
pageSubscription = null;
}
- return new QueueConfig(id, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, exclusive, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, purgeOnNoConsumers, groupRebalance, groupBuckets, configurationManaged);
+ return new QueueConfig(id, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, exclusive, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, purgeOnNoConsumers, groupRebalance, groupBuckets, autoDelete, autoDeleteDelay, autoDeleteMessageCount, configurationManaged);
}
}
@@ -286,6 +311,9 @@ public final class QueueConfig {
final boolean purgeOnNoConsumers,
final boolean groupRebalance,
final int groupBuckets,
+ final boolean autoDelete,
+ final long autoDeleteDelay,
+ final long autoDeleteMessageCount,
final boolean configurationManaged) {
this.id = id;
this.address = address;
@@ -307,6 +335,9 @@ public final class QueueConfig {
this.delayBeforeDispatch = delayBeforeDispatch;
this.groupRebalance = groupRebalance;
this.groupBuckets = groupBuckets;
+ this.autoDelete = autoDelete;
+ this.autoDeleteDelay = autoDeleteDelay;
+ this.autoDeleteMessageCount = autoDeleteMessageCount;
this.configurationManaged = configurationManaged;
}
@@ -394,6 +425,18 @@ public final class QueueConfig {
return configurationManaged;
}
+ public boolean isAutoDelete() {
+ return autoDelete;
+ }
+
+ public long getAutoDeleteDelay() {
+ return autoDeleteDelay;
+ }
+
+ public long getAutoDeleteMessageCount() {
+ return autoDeleteMessageCount;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o)
@@ -443,6 +486,12 @@ public final class QueueConfig {
return false;
if (groupBuckets != that.groupBuckets)
return false;
+ if (autoDelete != that.autoDelete)
+ return false;
+ if (autoDeleteDelay != that.autoDeleteDelay)
+ return false;
+ if (autoDeleteMessageCount != that.autoDeleteMessageCount)
+ return false;
if (configurationManaged != that.configurationManaged)
return false;
return user != null ? user.equals(that.user) : that.user == null;
@@ -471,6 +520,9 @@ public final class QueueConfig {
result = 31 * result + (purgeOnNoConsumers ? 1 : 0);
result = 31 * result + (groupRebalance ? 1 : 0);
result = 31 * result + groupBuckets;
+ result = 31 * result + (autoDelete ? 1 : 0);
+ result = 31 * result + Long.hashCode(autoDeleteDelay);
+ result = 31 * result + Long.hashCode(autoDeleteMessageCount);
result = 31 * result + (configurationManaged ? 1 : 0);
return result;
}
@@ -498,6 +550,9 @@ public final class QueueConfig {
+ ", purgeOnNoConsumers=" + purgeOnNoConsumers
+ ", groupRebalance=" + groupRebalance
+ ", groupBuckets=" + groupBuckets
+ + ", autoDelete=" + autoDelete
+ + ", autoDeleteDelay=" + autoDeleteDelay
+ + ", autoDeleteMessageCount=" + autoDeleteMessageCount
+ ", configurationManaged=" + configurationManaged + '}';
}
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerSession.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerSession.java
index 6ba9b3f..6f3708e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerSession.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerSession.java
@@ -188,6 +188,9 @@ public interface ServerSession extends SecurityAuth {
Boolean nonDestructive,
Integer consumersBeforeDispatch,
Long delayBeforeDispatch,
+ Boolean autoDelete,
+ Long autoDeleteDelay,
+ Long autoDeleteMessageCount,
boolean autoCreated) throws Exception;
Queue createQueue(SimpleString address,
@@ -343,7 +346,10 @@ public interface ServerSession extends SecurityAuth {
SimpleString lastValueKey,
Boolean nonDestructive,
Integer consumersBeforeDispatch,
- Long delayBeforeDispatch) throws Exception;
+ Long delayBeforeDispatch,
+ Boolean autoDelete,
+ Long autoDeleteDelay,
+ Long autoDeleteMessageCount) throws Exception;
void createSharedQueue(SimpleString address,
SimpleString name,
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 61600cb..adfb5ff 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
@@ -911,6 +911,9 @@ public class ActiveMQServerImpl implements ActiveMQServer {
int defaultConsumerWindowSize = addressSettings.getDefaultConsumerWindowSize();
boolean defaultGroupRebalance = addressSettings.isDefaultGroupRebalance();
int defaultGroupBuckets = addressSettings.getDefaultGroupBuckets();
+ boolean autoDeleteQueues = addressSettings.isAutoDeleteQueues();
+ long autoDeleteQueuesDelay = addressSettings.getAutoDeleteQueuesDelay();
+ long autoDeleteQueuesMessageCount = addressSettings.getAutoDeleteQueuesMessageCount();
SimpleString managementAddress = getManagementService() != null ? getManagementService().getManagementAddress() : null;
@@ -921,12 +924,12 @@ public class ActiveMQServerImpl implements ActiveMQServer {
SimpleString filterString = filter == null ? null : filter.getFilterString();
- response = new QueueQueryResult(realName, binding.getAddress(), queue.isDurable(), queue.isTemporary(), filterString, queue.getConsumerCount(), queue.getMessageCount(), autoCreateQueues, true, queue.isAutoCreated(), queue.isPurgeOnNoConsumers(), queue.getRoutingType(), queue.getMaxConsumers(), queue.isExclusive(), queue.isGroupRebalance(), queue.getGroupBuckets(), queue.isLastValue(), queue.getLastValueKey(), queue.isNonDestructive(), queue.getConsumersBeforeDispatch(), queue.ge [...]
+ response = new QueueQueryResult(realName, binding.getAddress(), queue.isDurable(), queue.isTemporary(), filterString, queue.getConsumerCount(), queue.getMessageCount(), autoCreateQueues, true, queue.isAutoCreated(), queue.isPurgeOnNoConsumers(), queue.getRoutingType(), queue.getMaxConsumers(), queue.isExclusive(), queue.isGroupRebalance(), queue.getGroupBuckets(), queue.isLastValue(), queue.getLastValueKey(), queue.isNonDestructive(), queue.getConsumersBeforeDispatch(), queue.ge [...]
} else if (realName.equals(managementAddress)) {
// make an exception for the management address (see HORNETQ-29)
- response = new QueueQueryResult(realName, managementAddress, true, false, null, -1, -1, autoCreateQueues, true, false, false, RoutingType.MULTICAST, -1, false, false, null, null, null, null, null, null, defaultConsumerWindowSize);
+ response = new QueueQueryResult(realName, managementAddress, true, false, null, -1, -1, autoCreateQueues, true, false, false, RoutingType.MULTICAST, -1, false, false, null, null, null, null, null, null, null, null, null, defaultConsumerWindowSize);
} else {
- response = new QueueQueryResult(realName, addressName, true, false, null, 0, 0, autoCreateQueues, false, false, defaultPurgeOnNoConsumers, RoutingType.MULTICAST, defaultMaxConsumers, defaultExclusiveQueue, defaultGroupRebalance, defaultGroupBuckets, defaultLastValueQueue, defaultLastValueKey, defaultNonDestructive, defaultConsumersBeforeDispatch, defaultDelayBeforeDispatch, defaultConsumerWindowSize);
+ response = new QueueQueryResult(realName, addressName, true, false, null, 0, 0, autoCreateQueues, false, false, defaultPurgeOnNoConsumers, RoutingType.MULTICAST, defaultMaxConsumers, defaultExclusiveQueue, defaultGroupRebalance, defaultGroupBuckets, defaultLastValueQueue, defaultLastValueKey, defaultNonDestructive, defaultConsumersBeforeDispatch, defaultDelayBeforeDispatch, autoDeleteQueues, autoDeleteQueuesDelay, autoDeleteQueuesMessageCount, defaultConsumerWindowSize);
}
return response;
@@ -1721,8 +1724,11 @@ public class ActiveMQServerImpl implements ActiveMQServer {
final boolean nonDestructive,
final int consumersBeforeDispatch,
final long delayBeforeDispatch,
+ final boolean autoDelete,
+ final long autoDeleteDelay,
+ final long autoDeleteMessageCount,
final boolean autoCreateAddress) throws Exception {
- return createQueue(address, routingType, queueName, filter, null, durable, temporary, false, false, false, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoCreateAddress);
+ return createQueue(address, routingType, queueName, filter, null, durable, temporary, false, false, false, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoDelete, autoDeleteDelay, autoDeleteMessageCount, autoCreateAddress);
}
@Override
@@ -1744,18 +1750,18 @@ public class ActiveMQServerImpl implements ActiveMQServer {
@Override
public Queue createQueue(AddressInfo addressInfo, SimpleString queueName, SimpleString filter, SimpleString user, boolean durable, boolean temporary, boolean autoCreated, Integer maxConsumers, Boolean purgeOnNoConsumers, boolean autoCreateAddress) throws Exception {
AddressSettings as = getAddressSettingsRepository().getMatch(addressInfo == null ? queueName.toString() : addressInfo.getName().toString());
- return createQueue(addressInfo, queueName, filter, user, durable, temporary, false, false, autoCreated, maxConsumers, purgeOnNoConsumers, as.isDefaultExclusiveQueue(), as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), as.isDefaultLastValueQueue(), as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), autoCreateAddress, false);
+ return createQueue(addressInfo, queueName, filter, user, durable, temporary, false, false, autoCreated, maxConsumers, purgeOnNoConsumers, as.isDefaultExclusiveQueue(), as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), as.isDefaultLastValueQueue(), as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), as.isAutoDeleteQueues(), as.getAutoDeleteQueuesDelay(), as.getAutoDeleteQueuesMessageCount(), [...]
}
@Override
public Queue createQueue(AddressInfo addressInfo, SimpleString queueName, SimpleString filter, SimpleString user, boolean durable, boolean temporary, boolean autoCreated, Integer maxConsumers, Boolean purgeOnNoConsumers, Boolean exclusive, Boolean lastValue, boolean autoCreateAddress) throws Exception {
AddressSettings as = getAddressSettingsRepository().getMatch(addressInfo == null ? queueName.toString() : addressInfo.getName().toString());
- return createQueue(addressInfo, queueName, filter, user, durable, temporary, false, false, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), lastValue, as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), autoCreateAddress, false);
+ return createQueue(addressInfo, queueName, filter, user, durable, temporary, false, false, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), lastValue, as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), as.isAutoDeleteQueues(), as.getAutoDeleteQueuesDelay(), as.getAutoDeleteQueuesMessageCount(), autoCreateAddress, false);
}
@Override
- public Queue createQueue(AddressInfo addressInfo, SimpleString queueName, SimpleString filter, SimpleString user, boolean durable, boolean temporary, boolean autoCreated, Integer maxConsumers, Boolean purgeOnNoConsumers, Boolean exclusive, Boolean groupRebalance, Integer groupBuckets, Boolean lastValue, SimpleString lastValueKey, Boolean nonDestructive, Integer consumersBeforeDispatch, Long delayBeforeDispatch, boolean autoCreateAddress) throws Exception {
- return createQueue(addressInfo, queueName, filter, user, durable, temporary, false, false, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoCreateAddress, false);
+ public Queue createQueue(AddressInfo addressInfo, SimpleString queueName, SimpleString filter, SimpleString user, boolean durable, boolean temporary, boolean autoCreated, Integer maxConsumers, Boolean purgeOnNoConsumers, Boolean exclusive, Boolean groupRebalance, Integer groupBuckets, Boolean lastValue, SimpleString lastValueKey, Boolean nonDestructive, Integer consumersBeforeDispatch, Long delayBeforeDispatch, Boolean autoDelete, Long autoDeleteDelay, Long autoDeleteMessageCount, boo [...]
+ return createQueue(addressInfo, queueName, filter, user, durable, temporary, false, false, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoDelete, autoDeleteDelay, autoDeleteMessageCount, autoCreateAddress, false);
}
@@ -1764,7 +1770,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
SimpleString user, boolean durable, boolean temporary, boolean ignoreIfExists, boolean transientQueue,
boolean autoCreated, int maxConsumers, boolean purgeOnNoConsumers, boolean autoCreateAddress) throws Exception {
AddressSettings as = getAddressSettingsRepository().getMatch(address == null ? queueName.toString() : address.toString());
- return createQueue(address, routingType, queueName, filter, user, durable, temporary, ignoreIfExists, transientQueue, autoCreated, maxConsumers, purgeOnNoConsumers, as.isDefaultExclusiveQueue(), as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), as.isDefaultLastValueQueue(), as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), autoCreateAddress);
+ return createQueue(address, routingType, queueName, filter, user, durable, temporary, ignoreIfExists, transientQueue, autoCreated, maxConsumers, purgeOnNoConsumers, as.isDefaultExclusiveQueue(), as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), as.isDefaultLastValueQueue(), as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), as.isAutoDeleteQueues(), as.getAutoDeleteQueuesDelay(), as.getAutoD [...]
}
@Override
@@ -1772,7 +1778,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
SimpleString user, boolean durable, boolean temporary, boolean ignoreIfExists, boolean transientQueue,
boolean autoCreated, int maxConsumers, boolean purgeOnNoConsumers, boolean exclusive, boolean lastValue, boolean autoCreateAddress) throws Exception {
AddressSettings as = getAddressSettingsRepository().getMatch(address == null ? queueName.toString() : address.toString());
- return createQueue(address, routingType, queueName, filter, user, durable, temporary, ignoreIfExists, transientQueue, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), lastValue, as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), autoCreateAddress);
+ return createQueue(address, routingType, queueName, filter, user, durable, temporary, ignoreIfExists, transientQueue, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), lastValue, as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), as.isAutoDeleteQueues(), as.getAutoDeleteQueuesDelay(), as.getAutoDeleteQueuesMessageCount(), autoCreate [...]
}
@@ -1809,7 +1815,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
boolean exclusive,
boolean lastValue) throws Exception {
AddressSettings as = getAddressSettingsRepository().getMatch(address == null ? name.toString() : address.toString());
- createSharedQueue(address, routingType, name, filterString, user, durable, maxConsumers, purgeOnNoConsumers, exclusive, as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), lastValue, as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch());
+ createSharedQueue(address, routingType, name, filterString, user, durable, maxConsumers, purgeOnNoConsumers, exclusive, as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), lastValue, as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), as.isAutoDeleteQueues(), as.getAutoDeleteQueuesDelay(), as.getAutoDeleteQueuesMessageCount());
}
@Override
@@ -1828,7 +1834,10 @@ public class ActiveMQServerImpl implements ActiveMQServer {
SimpleString lastValueKey,
boolean nonDestructive,
int consumersBeforeDispatch,
- long delayBeforeDispatch) throws Exception {
+ long delayBeforeDispatch,
+ boolean autoDelete,
+ long autoDeleteDelay,
+ long autoDeleteMessageCount) throws Exception {
//force the old contract about address
if (address == null) {
throw new NullPointerException("address can't be null!");
@@ -1842,7 +1851,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
}
}
- final Queue queue = createQueue(address, routingType, name, filterString, user, durable, !durable, true, !durable, false, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, true);
+ final Queue queue = createQueue(address, routingType, name, filterString, user, durable, !durable, true, !durable, false, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoDelete, autoDeleteDelay, autoDeleteMessageCount, true);
if (!queue.getAddress().equals(address)) {
throw ActiveMQMessageBundle.BUNDLE.queueSubscriptionBelongsToDifferentAddress(name);
@@ -1988,9 +1997,11 @@ public class ActiveMQServerImpl implements ActiveMQServer {
}
// This check is only valid if checkMessageCount == true
- long messageCount = queue.getMessageCount();
- if (checkMessageCount && messageCount != 0) {
- throw ActiveMQMessageBundle.BUNDLE.cannotDeleteQueueWithMessages(queue.getName(), queueName, messageCount);
+ if (checkMessageCount && queue.getAutoDeleteMessageCount() != -1) {
+ long messageCount = queue.getMessageCount();
+ if (queue.getMessageCount() > queue.getAutoDeleteMessageCount()) {
+ throw ActiveMQMessageBundle.BUNDLE.cannotDeleteQueueWithMessages(queue.getName(), queueName, messageCount);
+ }
}
queue.deleteQueue(removeConsumers);
@@ -2856,7 +2867,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
} else {
// if the address::queue doesn't exist then create it
try {
- createQueue(new AddressInfo(SimpleString.toSimpleString(config.getAddress())).addRoutingType(config.getRoutingType()), queueName, SimpleString.toSimpleString(config.getFilterString()), SimpleString.toSimpleString(config.getUser()), config.isDurable(), false, false, false, false, maxConsumers, config.getPurgeOnNoConsumers(), isExclusive, groupRebalance, groupBuckets, isLastValue, lastValueKey, isNonDestructive, consumersBeforeDispatch, delayBeforeDispatch, true, true);
+ createQueue(new AddressInfo(SimpleString.toSimpleString(config.getAddress())).addRoutingType(config.getRoutingType()), queueName, SimpleString.toSimpleString(config.getFilterString()), SimpleString.toSimpleString(config.getUser()), config.isDurable(), false, false, false, false, maxConsumers, config.getPurgeOnNoConsumers(), isExclusive, groupRebalance, groupBuckets, isLastValue, lastValueKey, isNonDestructive, consumersBeforeDispatch, delayBeforeDispatch, as.isAutoDelet [...]
} catch (ActiveMQQueueExistsException e) {
// the queue may exist on a *different* address
ActiveMQServerLogger.LOGGER.warn(e.getMessage());
@@ -3054,6 +3065,9 @@ public class ActiveMQServerImpl implements ActiveMQServer {
final boolean nonDestructive,
final int consumersBeforeDispatch,
final long delayBeforeDispatch,
+ final boolean autoDelete,
+ final long autoDeleteDelay,
+ final long autoDeleteMessageCount,
final boolean autoCreateAddress,
final boolean configurationManaged) throws Exception {
SimpleString realQueueName = CompositeAddress.extractQueueName(queueName);
@@ -3121,6 +3135,9 @@ public class ActiveMQServerImpl implements ActiveMQServer {
.nonDestructive(nonDestructive)
.consumersBeforeDispatch(consumersBeforeDispatch)
.delayBeforeDispatch(delayBeforeDispatch)
+ .autoDelete(autoDelete)
+ .autoDeleteDelay(autoDeleteDelay)
+ .autoDeleteMessageCount(autoDeleteMessageCount)
.configurationManaged(configurationManaged)
.build();
@@ -3200,8 +3217,11 @@ public class ActiveMQServerImpl implements ActiveMQServer {
final boolean nonDestructive,
final int consumersBeforeDispatch,
final long delayBeforeDispatch,
+ final boolean autoDelete,
+ final long autoDeleteDelay,
+ final long autoDeleteMessageCount,
final boolean autoCreateAddress) throws Exception {
- return createQueue(new AddressInfo(address).addRoutingType(routingType), queueName, filterString, user, durable, temporary, ignoreIfExists, transientQueue, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoCreateAddress, false);
+ return createQueue(new AddressInfo(address).addRoutingType(routingType), queueName, filterString, user, durable, temporary, ignoreIfExists, transientQueue, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoDelete, autoDeleteDelay, autoDeleteMessageCount, autoCreateAddress, false);
}
@Deprecated
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
index ee9e0e5..bd87b3e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
@@ -75,6 +75,9 @@ public class LastValueQueue extends QueueImpl {
final Boolean purgeOnNoConsumers,
final SimpleString lastValueKey,
final Boolean nonDestructive,
+ final Boolean autoDelete,
+ final Long autoDeleteDelay,
+ final Long autoDeleteMessageCount,
final boolean configurationManaged,
final ScheduledExecutorService scheduledExecutor,
final PostOffice postOffice,
@@ -83,7 +86,7 @@ public class LastValueQueue extends QueueImpl {
final ArtemisExecutor executor,
final ActiveMQServer server,
final QueueFactory factory) {
- super(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, exclusive, groupRebalance, groupBuckets, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, purgeOnNoConsumers, configurationManaged, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server, factory);
+ super(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, exclusive, groupRebalance, groupBuckets, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, purgeOnNoConsumers, autoDelete, autoDeleteDelay, autoDeleteMessageCount, configurationManaged, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server, factory);
this.lastValueKey = lastValueKey;
}
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 606256c..defb23b 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
@@ -159,6 +159,9 @@ public class PostOfficeJournalLoader implements JournalLoader {
.nonDestructive(queueBindingInfo.isNonDestructive())
.consumersBeforeDispatch(queueBindingInfo.getConsumersBeforeDispatch())
.delayBeforeDispatch(queueBindingInfo.getDelayBeforeDispatch())
+ .autoDelete(queueBindingInfo.isAutoDelete())
+ .autoDeleteDelay(queueBindingInfo.getAutoDeleteDelay())
+ .autoDeleteMessageCount(queueBindingInfo.getAutoDeleteMessageCount())
.routingType(RoutingType.getType(queueBindingInfo.getRoutingType()))
.configurationManaged((queueBindingInfo.isConfigurationManaged()));
final Queue queue = queueFactory.createQueueWith(queueConfigBuilder.build());
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
index cba671a..f57d391 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java
@@ -75,9 +75,9 @@ public class QueueFactoryImpl implements QueueFactory {
public Queue createQueueWith(final QueueConfig config) {
final Queue queue;
if (lastValueKey(config) != null) {
- queue = new LastValueQueue(config.id(), config.address(), config.name(), config.filter(), config.pageSubscription(), config.user(), config.isDurable(), config.isTemporary(), config.isAutoCreated(), config.deliveryMode(), config.maxConsumers(), config.isExclusive(), config.isGroupRebalance(), config.getGroupBuckets(), config.consumersBeforeDispatch(), config.delayBeforeDispatch(), config.isPurgeOnNoConsumers(), lastValueKey(config), config.isNonDestructive(), config.isConfigurati [...]
+ queue = new LastValueQueue(config.id(), config.address(), config.name(), config.filter(), config.pageSubscription(), config.user(), config.isDurable(), config.isTemporary(), config.isAutoCreated(), config.deliveryMode(), config.maxConsumers(), config.isExclusive(), config.isGroupRebalance(), config.getGroupBuckets(), config.consumersBeforeDispatch(), config.delayBeforeDispatch(), config.isPurgeOnNoConsumers(), lastValueKey(config), config.isNonDestructive(), config.isAutoDelete( [...]
} else {
- queue = new QueueImpl(config.id(), config.address(), config.name(), config.filter(), config.pageSubscription(), config.user(), config.isDurable(), config.isTemporary(), config.isAutoCreated(), config.deliveryMode(), config.maxConsumers(), config.isExclusive(), config.isGroupRebalance(), config.getGroupBuckets(), config.isNonDestructive(), config.consumersBeforeDispatch(), config.delayBeforeDispatch(), config.isPurgeOnNoConsumers(), config.isConfigurationManaged(), scheduledExecu [...]
+ queue = new QueueImpl(config.id(), config.address(), config.name(), config.filter(), config.pageSubscription(), config.user(), config.isDurable(), config.isTemporary(), config.isAutoCreated(), config.deliveryMode(), config.maxConsumers(), config.isExclusive(), config.isGroupRebalance(), config.getGroupBuckets(), config.isNonDestructive(), config.consumersBeforeDispatch(), config.delayBeforeDispatch(), config.isPurgeOnNoConsumers(), config.isAutoDelete(), config.getAutoDeleteDela [...]
}
server.getCriticalAnalyzer().add(queue);
return queue;
@@ -102,7 +102,7 @@ public class QueueFactoryImpl implements QueueFactory {
Queue queue;
if (lastValueKey(addressSettings) != null) {
- queue = new LastValueQueue(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, ActiveMQDefaultConfiguration.getDefaultRoutingType(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), ActiveMQDefaultConfiguration.getDefaultExclusive(), ActiveMQDefaultConfiguration.getDefaultGroupRebalance(), ActiveMQDefaultConfiguration.getDefaultGroupBuckets(), ActiveMQDefaultConfiguration.getDefaultConsumersBeforeDispatch(), ActiveMQDefaultCo [...]
+ queue = new LastValueQueue(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, ActiveMQDefaultConfiguration.getDefaultRoutingType(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), ActiveMQDefaultConfiguration.getDefaultExclusive(), ActiveMQDefaultConfiguration.getDefaultGroupRebalance(), ActiveMQDefaultConfiguration.getDefaultGroupBuckets(), ActiveMQDefaultConfiguration.getDefaultConsumersBeforeDispatch(), ActiveMQDefaultCo [...]
} else {
queue = new QueueImpl(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this);
}
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 31e166e..79e53a9 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
@@ -285,6 +285,12 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
private volatile long delayBeforeDispatch = 0;
+ private final boolean autoDelete;
+
+ private final long autoDeleteDelay;
+
+ private final long autoDeleteMessageCount;
+
private volatile boolean configurationManaged;
private volatile boolean nonDestructive;
@@ -411,7 +417,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
final ArtemisExecutor executor,
final ActiveMQServer server,
final QueueFactory factory) {
- this(id, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, exclusive, null, null, false, null, null, purgeOnNoConsumers, false, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server, factory);
+ this(id, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, exclusive, null, null, false, null, null, purgeOnNoConsumers, null, null, null, false, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server, factory);
}
public QueueImpl(final long id,
@@ -432,6 +438,9 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
final Integer consumersBeforeDispatch,
final Long delayBeforeDispatch,
final Boolean purgeOnNoConsumers,
+ final Boolean autoDelete,
+ final Long autoDeleteDelay,
+ final Long autoDeleteMessageCount,
final boolean configurationManaged,
final ScheduledExecutorService scheduledExecutor,
final PostOffice postOffice,
@@ -480,6 +489,12 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
this.groups = groupMap(this.groupBuckets);
+ this.autoDelete = autoDelete == null ? ActiveMQDefaultConfiguration.getDefaultQueueAutoDelete() : autoDelete;
+
+ this.autoDeleteDelay = autoDeleteDelay == null ? ActiveMQDefaultConfiguration.getDefaultQueueAutoDeleteDelay() : autoDeleteDelay;
+
+ this.autoDeleteMessageCount = autoDeleteMessageCount == null ? ActiveMQDefaultConfiguration.getDefaultQueueAutoDeleteMessageCount() : autoDeleteMessageCount;
+
this.configurationManaged = configurationManaged;
this.postOffice = postOffice;
@@ -657,6 +672,21 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
}
@Override
+ public boolean isAutoDelete() {
+ return autoDelete;
+ }
+
+ @Override
+ public long getAutoDeleteDelay() {
+ return autoDeleteDelay;
+ }
+
+ @Override
+ public long getAutoDeleteMessageCount() {
+ return autoDeleteMessageCount;
+ }
+
+ @Override
public boolean isTemporary() {
return temporary;
}
@@ -1980,6 +2010,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
if (queueDestroyed) {
return;
}
+
if (logger.isDebugEnabled()) {
logger.debug("Scanning for expires on " + QueueImpl.this.getName());
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueManagerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueManagerImpl.java
index dd34bdc..08fcddd 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueManagerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueManagerImpl.java
@@ -40,30 +40,56 @@ public class QueueManagerImpl extends ReferenceCounterUtil implements QueueManag
return;
}
- AddressSettings settings = server.getAddressSettingsRepository().getMatch(queue.getAddress().toString());
+ if (isAutoDelete(queue) && consumerCountCheck(queue) && delayCheck(queue) && messageCountCheck(queue)) {
+ deleteAutoCreatedQueue(server, queue);
+ } else if (queue.isPurgeOnNoConsumers()) {
+ purge(queue);
+ }
+ }
+
+ private static void purge(Queue queue) {
long consumerCount = queue.getConsumerCount();
long messageCount = queue.getMessageCount();
- if (queue.isAutoCreated() && settings.isAutoDeleteQueues() && queue.getMessageCount() == 0 && queue.getConsumerCount() == 0 && settings.getAutoDeleteQueuesDelay() == 0) {
- if (ActiveMQServerLogger.LOGGER.isDebugEnabled()) {
- ActiveMQServerLogger.LOGGER.debug("deleting " + (queue.isAutoCreated() ? "auto-created " : "") + "queue \"" + queueName + ".\" consumerCount = " + consumerCount + "; messageCount = " + messageCount + "; isAutoDeleteQueues = " + settings.isAutoDeleteQueues());
- }
+ if (ActiveMQServerLogger.LOGGER.isDebugEnabled()) {
+ ActiveMQServerLogger.LOGGER.debug("purging queue \"" + queue.getName() + ".\" consumerCount = " + consumerCount + "; messageCount = " + messageCount);
+ }
+ try {
+ queue.deleteMatchingReferences(QueueImpl.DEFAULT_FLUSH_LIMIT, null, AckReason.KILLED);
+ } catch (Exception e) {
+ ActiveMQServerLogger.LOGGER.failedToPurgeQueue(e, queue.getName());
+ }
+ }
- try {
- server.destroyQueue(queueName, null, true, false, settings.isAutoDeleteAddresses(), true);
- } catch (Exception e) {
- ActiveMQServerLogger.LOGGER.errorRemovingAutoCreatedQueue(e, queueName);
- }
- } else if (queue.isPurgeOnNoConsumers()) {
- if (ActiveMQServerLogger.LOGGER.isDebugEnabled()) {
- ActiveMQServerLogger.LOGGER.debug("purging queue \"" + queueName + ".\" consumerCount = " + consumerCount + "; messageCount = " + messageCount);
- }
- try {
- queue.deleteMatchingReferences(QueueImpl.DEFAULT_FLUSH_LIMIT, null, AckReason.KILLED);
- } catch (Exception e) {
- ActiveMQServerLogger.LOGGER.failedToPurgeQueue(e, queueName);
- }
+ public static void deleteAutoCreatedQueue(ActiveMQServer server, Queue queue) {
+ SimpleString queueName = queue.getName();
+ AddressSettings settings = server.getAddressSettingsRepository().getMatch(queue.getAddress().toString());
+ if (ActiveMQServerLogger.LOGGER.isDebugEnabled()) {
+ ActiveMQServerLogger.LOGGER.info("deleting auto-created queue \"" + queueName + ".\" consumerCount = " + queue.getConsumerCount() + "; messageCount = " + queue.getMessageCount() + "; isAutoDeleteQueues = " + settings.isAutoDeleteQueues());
}
+
+ try {
+ server.destroyQueue(queueName, null, true, false, settings.isAutoDeleteAddresses(), true);
+ } catch (Exception e) {
+ ActiveMQServerLogger.LOGGER.errorRemovingAutoCreatedQueue(e, queueName);
+ }
+ }
+
+ public static boolean isAutoDelete(Queue queue) {
+ return queue.isAutoCreated() && queue.isAutoDelete();
+ }
+
+ public static boolean messageCountCheck(Queue queue) {
+ return queue.getAutoDeleteMessageCount() == -1 || queue.getMessageCount() <= queue.getAutoDeleteMessageCount();
+ }
+
+ public static boolean delayCheck(Queue queue) {
+ long consumerRemovedTimestamp = queue.getConsumerRemovedTimestamp();
+ return consumerRemovedTimestamp != -1 && System.currentTimeMillis() - consumerRemovedTimestamp >= queue.getAutoDeleteDelay();
+ }
+
+ public static boolean consumerCountCheck(Queue queue) {
+ return queue.getConsumerCount() == 0;
}
public QueueManagerImpl(ActiveMQServer server, SimpleString queueName) {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
index fd26db4..6464eee 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
@@ -591,7 +591,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
@Override
public Queue createQueue(AddressInfo addressInfo, SimpleString name, SimpleString filterString, boolean temporary, boolean durable) throws Exception {
AddressSettings as = server.getAddressSettingsRepository().getMatch(addressInfo.getName().toString());
- return createQueue(addressInfo, name, filterString, temporary, durable, as.getDefaultMaxConsumers(), as.isDefaultPurgeOnNoConsumers(), as.isDefaultExclusiveQueue(), as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), as.isDefaultLastValueQueue(), as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), false);
+ return createQueue(addressInfo, name, filterString, temporary, durable, as.getDefaultMaxConsumers(), as.isDefaultPurgeOnNoConsumers(), as.isDefaultExclusiveQueue(), as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), as.isDefaultLastValueQueue(), as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), as.isAutoDeleteQueues(), as.getAutoDeleteQueuesDelay(), as.getAutoDeleteQueuesMessageCount(), false);
}
public Queue createQueue(final AddressInfo addressInfo,
@@ -609,6 +609,9 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
final boolean nonDestructive,
final int consumersBeforeDispatch,
final long delayBeforeDispatch,
+ final boolean autoDelete,
+ final long autoDeleteDelay,
+ final long autoDeleteMessageCount,
final boolean autoCreated) throws Exception {
final SimpleString unPrefixedName = removePrefix(name);
@@ -629,7 +632,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
server.checkQueueCreationLimit(getUsername());
- Queue queue = server.createQueue(art, unPrefixedName, filterString, SimpleString.toSimpleString(getUsername()), durable, temporary, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, as.isAutoCreateAddresses());
+ Queue queue = server.createQueue(art, unPrefixedName, filterString, SimpleString.toSimpleString(getUsername()), durable, temporary, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoDelete, autoDeleteDelay, autoDeleteMessageCount, as.isAutoCreateAddresses());
if (temporary) {
// Temporary queue in core simply means the queue will be deleted if
@@ -669,7 +672,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
final boolean purgeOnNoConsumers,
final boolean autoCreated) throws Exception {
AddressSettings as = server.getAddressSettingsRepository().getMatch(address.toString());
- return createQueue(new AddressInfo(address, routingType), name, filterString, temporary, durable, maxConsumers, purgeOnNoConsumers, as.isDefaultExclusiveQueue(), as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), as.isDefaultLastValueQueue(), as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), autoCreated);
+ return createQueue(new AddressInfo(address, routingType), name, filterString, temporary, durable, maxConsumers, purgeOnNoConsumers, as.isDefaultExclusiveQueue(), as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), as.isDefaultLastValueQueue(), as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), as.isAutoDeleteQueues(), as.getAutoDeleteQueuesDelay(), as.getAutoDeleteQueuesMessageCount(), autoCreated);
}
@Override
@@ -684,7 +687,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
final Boolean exclusive,
final Boolean lastValue,
final boolean autoCreated) throws Exception {
- return createQueue(address, name, routingType, filterString, temporary, durable, maxConsumers, purgeOnNoConsumers, exclusive, null, null, lastValue, null, null, null, null, autoCreated);
+ return createQueue(address, name, routingType, filterString, temporary, durable, maxConsumers, purgeOnNoConsumers, exclusive, null, null, lastValue, null, null, null, null, null, null, null, autoCreated);
}
@Override
@@ -704,8 +707,11 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
final Boolean nonDestructive,
final Integer consumersBeforeDispatch,
final Long delayBeforeDispatch,
+ final Boolean autoDelete,
+ final Long autoDeleteDelay,
+ final Long autoDeleteMessageCount,
final boolean autoCreated) throws Exception {
- if (exclusive == null || groupRebalance == null || groupBuckets == null || lastValue == null || lastValueKey == null || nonDestructive == null || consumersBeforeDispatch == null || delayBeforeDispatch == null) {
+ if (exclusive == null || groupRebalance == null || groupBuckets == null || lastValue == null || lastValueKey == null || nonDestructive == null || consumersBeforeDispatch == null || delayBeforeDispatch == null || autoDelete == null || autoDeleteDelay == null || autoDeleteMessageCount == null) {
AddressSettings as = server.getAddressSettingsRepository().getMatch(address.toString());
return createQueue(new AddressInfo(address, routingType), name, filterString, temporary, durable, maxConsumers, purgeOnNoConsumers,
exclusive == null ? as.isDefaultExclusiveQueue() : exclusive,
@@ -716,10 +722,13 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
nonDestructive == null ? as.isDefaultNonDestructive() : nonDestructive,
consumersBeforeDispatch == null ? as.getDefaultConsumersBeforeDispatch() : consumersBeforeDispatch,
delayBeforeDispatch == null ? as.getDefaultDelayBeforeDispatch() : delayBeforeDispatch,
+ autoDelete == null ? as.isAutoDeleteQueues() : autoDelete,
+ autoDeleteDelay == null ? as.getAutoDeleteQueuesDelay() : autoDeleteDelay,
+ autoDeleteMessageCount == null ? as.getAutoDeleteQueuesMessageCount() : autoDeleteMessageCount,
autoCreated);
} else {
return createQueue(new AddressInfo(address, routingType), name, filterString, temporary, durable, maxConsumers, purgeOnNoConsumers,
- exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoCreated);
+ exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoDelete, autoDeleteDelay, autoDeleteMessageCount, autoCreated);
}
}
@@ -738,14 +747,14 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
@Override
public Queue createQueue(AddressInfo addressInfo, SimpleString name, SimpleString filterString, boolean temporary, boolean durable, boolean autoCreated) throws Exception {
AddressSettings as = server.getAddressSettingsRepository().getMatch(addressInfo.getName().toString());
- return createQueue(addressInfo, name, filterString, temporary, durable, as.getDefaultMaxConsumers(), as.isDefaultPurgeOnNoConsumers(), as.isDefaultExclusiveQueue(), as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), as.isDefaultLastValueQueue(), as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), autoCreated);
+ return createQueue(addressInfo, name, filterString, temporary, durable, as.getDefaultMaxConsumers(), as.isDefaultPurgeOnNoConsumers(), as.isDefaultExclusiveQueue(), as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), as.isDefaultLastValueQueue(), as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), as.isAutoDeleteQueues(), as.getAutoDeleteQueuesDelay(), as.getAutoDeleteQueuesMessageCount(), aut [...]
}
@Override
public Queue createQueue(AddressInfo addressInfo, SimpleString name, SimpleString filterString, boolean temporary, boolean durable, Boolean exclusive, Boolean lastValue, boolean autoCreated) throws Exception {
AddressSettings as = server.getAddressSettingsRepository().getMatch(addressInfo.getName().toString());
return createQueue(addressInfo, name, filterString, temporary, durable, as.getDefaultMaxConsumers(), as.isDefaultPurgeOnNoConsumers(),
- exclusive == null ? as.isDefaultExclusiveQueue() : exclusive, as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), lastValue == null ? as.isDefaultLastValueQueue() : lastValue, as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), autoCreated);
+ exclusive == null ? as.isDefaultExclusiveQueue() : exclusive, as.isDefaultGroupRebalance(), as.getDefaultGroupBuckets(), lastValue == null ? as.isDefaultLastValueQueue() : lastValue, as.getDefaultLastValueKey(), as.isDefaultNonDestructive(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch(), as.isAutoDeleteQueues(), as.getAutoDeleteQueuesDelay(), as.getAutoDeleteQueuesMessageCount(), autoCreated);
}
@Override
@@ -784,7 +793,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
Boolean purgeOnNoConsumers,
Boolean exclusive,
Boolean lastValue) throws Exception {
- createSharedQueue(address, name, routingType, filterString, durable, maxConsumers, purgeOnNoConsumers, exclusive, null, null, lastValue, null, null, null, null);
+ createSharedQueue(address, name, routingType, filterString, durable, maxConsumers, purgeOnNoConsumers, exclusive, null, null, lastValue, null, null, null, null, null, null, null);
}
@Override
@@ -802,7 +811,10 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
SimpleString lastValueKey,
Boolean nonDestructive,
Integer consumersBeforeDispatch,
- Long delayBeforeDispatch) throws Exception {
+ Long delayBeforeDispatch,
+ Boolean autoDelete,
+ Long autoDeleteDelay,
+ Long autoDeleteMessageCount) throws Exception {
address = removePrefix(address);
securityCheck(address, name, durable ? CheckType.CREATE_DURABLE_QUEUE : CheckType.CREATE_NON_DURABLE_QUEUE, this);
@@ -821,7 +833,10 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
lastValueKey == null ? as.getDefaultLastValueKey() : lastValueKey,
nonDestructive == null ? as.isDefaultNonDestructive() : nonDestructive,
consumersBeforeDispatch == null ? as.getDefaultConsumersBeforeDispatch() : consumersBeforeDispatch,
- delayBeforeDispatch == null ? as.getDefaultDelayBeforeDispatch() : delayBeforeDispatch);
+ delayBeforeDispatch == null ? as.getDefaultDelayBeforeDispatch() : delayBeforeDispatch,
+ autoDelete == null ? as.isAutoDeleteQueues() : autoDelete,
+ autoDeleteDelay == null ? as.getAutoDeleteQueuesDelay() : delayBeforeDispatch,
+ autoDeleteMessageCount == null ? as.getAutoDeleteQueuesMessageCount() : autoDeleteMessageCount);
}
@Override
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 2cb6b93..959ce75 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
@@ -74,6 +74,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
public static final long DEFAULT_AUTO_DELETE_QUEUES_DELAY = 0;
+ public static final long DEFAULT_AUTO_DELETE_QUEUES_MESSAGE_COUNT = 0;
+
public static final DeletionPolicy DEFAULT_CONFIG_DELETE_QUEUES = DeletionPolicy.OFF;
public static final boolean DEFAULT_AUTO_CREATE_ADDRESSES = true;
@@ -169,6 +171,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
private Long autoDeleteQueuesDelay = null;
+ private Long autoDeleteQueuesMessageCount = null;
+
private DeletionPolicy configDeleteQueues = null;
private Boolean autoCreateAddresses = null;
@@ -324,6 +328,16 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
return this;
}
+ public long getAutoDeleteQueuesMessageCount() {
+ return autoDeleteQueuesMessageCount != null ? autoDeleteQueuesMessageCount : AddressSettings.DEFAULT_AUTO_DELETE_QUEUES_MESSAGE_COUNT;
+ }
+
+ public AddressSettings setAutoDeleteQueuesMessageCount(final long autoDeleteQueuesMessageCount) {
+ this.autoDeleteQueuesMessageCount = autoDeleteQueuesMessageCount;
+ return this;
+ }
+
+
public DeletionPolicy getConfigDeleteQueues() {
return configDeleteQueues != null ? configDeleteQueues : AddressSettings.DEFAULT_CONFIG_DELETE_QUEUES;
}
@@ -768,6 +782,9 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
if (autoDeleteQueuesDelay == null) {
autoDeleteQueuesDelay = merged.autoDeleteQueuesDelay;
}
+ if (autoDeleteQueuesMessageCount == null) {
+ autoDeleteQueuesMessageCount = merged.autoDeleteQueuesMessageCount;
+ }
if (configDeleteQueues == null) {
configDeleteQueues = merged.configDeleteQueues;
}
@@ -978,6 +995,10 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
if (buffer.readableBytes() > 0) {
defaultGroupBuckets = BufferHelper.readNullableInteger(buffer);
}
+
+ if (buffer.readableBytes() > 0) {
+ autoDeleteQueuesMessageCount = BufferHelper.readNullableLong(buffer);
+ }
}
@Override
@@ -1025,7 +1046,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
BufferHelper.sizeOfNullableLong(autoDeleteQueuesDelay) +
BufferHelper.sizeOfNullableLong(autoDeleteAddressesDelay) +
BufferHelper.sizeOfNullableBoolean(defaultGroupRebalance) +
- BufferHelper.sizeOfNullableInteger(defaultGroupBuckets);
+ BufferHelper.sizeOfNullableInteger(defaultGroupBuckets) +
+ BufferHelper.sizeOfNullableLong(autoDeleteQueuesMessageCount);
}
@Override
@@ -1120,6 +1142,9 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
BufferHelper.writeNullableInteger(buffer, defaultGroupBuckets);
+ BufferHelper.writeNullableLong(buffer, autoDeleteQueuesMessageCount);
+
+
}
/* (non-Javadoc)
@@ -1158,6 +1183,7 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
result = prime * result + ((autoCreateQueues == null) ? 0 : autoCreateQueues.hashCode());
result = prime * result + ((autoDeleteQueues == null) ? 0 : autoDeleteQueues.hashCode());
result = prime * result + ((autoDeleteQueuesDelay == null) ? 0 : autoDeleteQueuesDelay.hashCode());
+ result = prime * result + ((autoDeleteQueuesMessageCount == null) ? 0 : autoDeleteQueuesMessageCount.hashCode());
result = prime * result + ((configDeleteQueues == null) ? 0 : configDeleteQueues.hashCode());
result = prime * result + ((autoCreateAddresses == null) ? 0 : autoCreateAddresses.hashCode());
result = prime * result + ((autoDeleteAddresses == null) ? 0 : autoDeleteAddresses.hashCode());
@@ -1335,6 +1361,11 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
return false;
} else if (!autoDeleteQueuesDelay.equals(other.autoDeleteQueuesDelay))
return false;
+ if (autoDeleteQueuesMessageCount == null) {
+ if (other.autoDeleteQueuesMessageCount != null)
+ return false;
+ } else if (!autoDeleteQueuesMessageCount.equals(other.autoDeleteQueuesMessageCount))
+ return false;
if (configDeleteQueues == null) {
if (other.configDeleteQueues != null)
return false;
@@ -1497,6 +1528,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
autoDeleteQueues +
", autoDeleteQueuesDelay=" +
autoDeleteQueuesDelay +
+ ", autoDeleteQueuesMessageCount=" +
+ autoDeleteQueuesMessageCount +
", configDeleteQueues=" +
configDeleteQueues +
", autoCreateAddresses=" +
diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
index ad46ed4..a93fe26 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -2991,11 +2991,18 @@
<xsd:annotation>
<xsd:documentation>
how long to wait (in milliseconds) before deleting auto-created queues after the queue has 0
- consumers and 0 messages
+ consumers.
</xsd:documentation>
</xsd:annotation>
</xsd:element>
+ <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>
+ </xsd:annotation>
+ </xsd:element>
+
<xsd:element name="config-delete-queues" default="OFF" maxOccurs="1" minOccurs="0">
<xsd:annotation>
<xsd:documentation>
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
index 1dc377a..40ab28e 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
@@ -947,6 +947,24 @@ public class ScheduledDeliveryHandlerTest extends Assert {
}
@Override
+ public boolean isAutoDelete() {
+ // no-op
+ return false;
+ }
+
+ @Override
+ public long getAutoDeleteDelay() {
+ // no-op
+ return -1;
+ }
+
+ @Override
+ public long getAutoDeleteMessageCount() {
+ // no-op
+ return -1;
+ }
+
+ @Override
public boolean isTemporary() {
return false;
}
diff --git a/docs/user-manual/en/address-model.md b/docs/user-manual/en/address-model.md
index 801e332..845ad59 100644
--- a/docs/user-manual/en/address-model.md
+++ b/docs/user-manual/en/address-model.md
@@ -598,6 +598,7 @@ that would be found in the `broker.xml` file.
<auto-create-queues>true</auto-create-queues>
<auto-delete-queues>true</auto-delete-queues>
<auto-delete-queues-delay>0</auto-delete-queues-delay>
+ <auto-delete-queues-message-count>0</auto-delete-queues-message-count>
<config-delete-queues>OFF</config-delete-queues>
<auto-create-addresses>true</auto-create-addresses>
<auto-delete-addresses>true</auto-delete-addresses>
@@ -760,15 +761,30 @@ name fits the address `match`. Queues which are auto-created are durable,
non-temporary, and non-transient. Default is `true`.
`auto-delete-queues`. Whether or not the broker should automatically delete
-auto-created queues when they have both 0 consumers and 0 messages. Default is
+auto-created queues when they have both 0 consumers and the message count is
+less than or equal to `auto-delete-queues-message-count`. Default is
`true`.
`auto-delete-queues-delay`. How long to wait (in milliseconds) before deleting
-auto-created queues after the queue has 0 consumers and 0 messages. Default is
-`0` (delete immediately). The broker's `address-queue-scan-period` controls
+auto-created queues after the queue has 0 consumers and the message count is
+less than or equal to `auto-delete-queues-message-count`.
+Default is `0` (delete immediately). The broker's `address-queue-scan-period` controls
how often (in milliseconds) queues are scanned for potential deletion. Use `-1`
to disable scanning. The default scan value is `30000`.
+`auto-delete-queues-message-count`. The message count that the queue must be
+less than or equal to before deleting auto-created queues.
+To disable message count check `-1` can be set.
+Default is `0` (empty queue).
+
+**Note:** the above auto-delete address settings can also be configured
+individually at the queue level when a client auto creates the queue.
+
+For Core API it is exposed in createQueue methods.
+
+For Core JMS you can set it using the destination queue attributes
+`my.destination?auto-delete=true&auto-delete-delay=120000&auto-delete-message-count=-1`
+
`config-delete-queues`. How the broker should handle queues deleted on config
reload, by delete policy: `OFF` or `FORCE`. Default is `OFF`. Read more about
[configuration reload](config-reload.md).
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/ConsumerDelayDispatchTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/ConsumerDelayDispatchTest.java
index c42eb29..51ccaf7 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/ConsumerDelayDispatchTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/ConsumerDelayDispatchTest.java
@@ -46,8 +46,8 @@ public class ConsumerDelayDispatchTest extends JMSTestBase {
@Before
public void setUp() throws Exception {
super.setUp();
- server.createQueue(queueName, RoutingType.ANYCAST, queueName, null, null, true, false, false, false, false, -1, false, true, false, -1, false, null, false, 2, DELAY_BEFORE_DISPATCH, true);
- server.createQueue(normalQueueName, RoutingType.ANYCAST, normalQueueName, null, null, true, false, false, false, false, -1, false, true, false, -1, false, null, false, 0, -1, true);
+ server.createQueue(queueName, RoutingType.ANYCAST, queueName, null, null, true, false, false, false, false, -1, false, true, false, -1, false, null, false, 2, DELAY_BEFORE_DISPATCH, false, 0, 0, true);
+ server.createQueue(normalQueueName, RoutingType.ANYCAST, normalQueueName, null, null, true, false, false, false, false, -1, false, true, false, -1, false, null, false, 0, -1, false, 0, 0, true);
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/GroupingTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/GroupingTest.java
index f8fa90c..cfd7b13 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/GroupingTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/GroupingTest.java
@@ -315,7 +315,7 @@ public class GroupingTest extends JMSTestBase {
String testQueueName = getName() + "_bucket_group";
- server.createQueue(SimpleString.toSimpleString(testQueueName), RoutingType.ANYCAST, SimpleString.toSimpleString(testQueueName), null, null, true, false, false, false, false, -1, false, false, false, 2, false, null, false, 0, 0, true);
+ server.createQueue(SimpleString.toSimpleString(testQueueName), RoutingType.ANYCAST, SimpleString.toSimpleString(testQueueName), null, null, true, false, false, false, false, -1, false, false, false, 2, false, null, false, 0, 0, false, 0, 0, true);
JMSContext ctx = addContext(getCF().createContext(JMSContext.SESSION_TRANSACTED));
@@ -397,7 +397,7 @@ public class GroupingTest extends JMSTestBase {
Assume.assumeTrue("only makes sense withOUT explicit group-id", ((ActiveMQConnectionFactory) fact).getGroupID() == null);
String testQueueName = getName() + "_group_rebalance";
- server.createQueue(SimpleString.toSimpleString(testQueueName), RoutingType.ANYCAST, SimpleString.toSimpleString(testQueueName), null, null, true, false, false, false, false, -1, false, false, true, -1, false, null, false, 0, 0, true);
+ server.createQueue(SimpleString.toSimpleString(testQueueName), RoutingType.ANYCAST, SimpleString.toSimpleString(testQueueName), null, null, true, false, false, false, false, -1, false, false, true, -1, false, null, false, 0, 0, false, 0, 0, true);
JMSContext ctx = addContext(getCF().createContext(JMSContext.SESSION_TRANSACTED));
@@ -527,7 +527,7 @@ public class GroupingTest extends JMSTestBase {
Assume.assumeTrue("only makes sense withOUT explicit group-id", ((ActiveMQConnectionFactory) fact).getGroupID() == null);
String testQueueName = getName() + "_group_disable";
- server.createQueue(SimpleString.toSimpleString(testQueueName), RoutingType.ANYCAST, SimpleString.toSimpleString(testQueueName), null, null, true, false, false, false, false, -1, false, false, false, 0, false, null, false, 0, 0, true);
+ server.createQueue(SimpleString.toSimpleString(testQueueName), RoutingType.ANYCAST, SimpleString.toSimpleString(testQueueName), null, null, true, false, false, false, false, -1, false, false, false, 0, false, null, false, 0, 0, false, 0, 0, true);
JMSContext ctx = addContext(getCF().createContext(JMSContext.SESSION_TRANSACTED));
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/QueueAutoDeleteTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/QueueAutoDeleteTest.java
new file mode 100644
index 0000000..4e7c091
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/QueueAutoDeleteTest.java
@@ -0,0 +1,325 @@
+/*
+ * 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.jms.client;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.postoffice.QueueBinding;
+import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
+import org.apache.activemq.artemis.tests.util.JMSTestBase;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * QueueAutoDeleteTest this tests that we can configure at the queue level auto-delete behaviour of auto created queues.
+ */
+public class QueueAutoDeleteTest extends JMSTestBase {
+
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ }
+
+ @Override
+ protected Configuration createDefaultConfig(boolean netty) throws Exception {
+ //Set scan period over aggressively so tests do not have to wait too long.
+ return super.createDefaultConfig(netty).setAddressQueueScanPeriod(10);
+ }
+
+ protected ConnectionFactory getCF() throws Exception {
+ return cf;
+ }
+
+ @Test
+ public void testAutoDelete() throws Exception {
+ ConnectionFactory fact = getCF();
+ Connection connection = fact.createConnection();
+ connection.start();
+
+ try {
+
+ Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+
+ String testQueueName = getName();
+
+ Queue queue = session.createQueue(testQueueName + "?auto-delete=true");
+ ActiveMQDestination activeMQDestination = (ActiveMQDestination) queue;
+
+ assertEquals(testQueueName, queue.getQueueName());
+ assertEquals(true, activeMQDestination.getQueueAttributes().getAutoDelete());
+
+ MessageProducer producer = session.createProducer(queue);
+ producer.send(session.createTextMessage("hello1"));
+ producer.send(session.createTextMessage("hello2"));
+
+ QueueBinding queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertTrue(queueBinding.getQueue().isAutoDelete());
+ assertEquals(2, queueBinding.getQueue().getMessageCount());
+
+ MessageConsumer consumer = session.createConsumer(queue);
+ Message message = consumer.receive(100);
+ assertNotNull(message);
+ message.acknowledge();
+
+ consumer.close();
+
+ queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertEquals(1, queueBinding.getQueue().getMessageCount());
+
+ consumer = session.createConsumer(queue);
+ message = consumer.receive(100);
+ assertNotNull(message);
+ message.acknowledge();
+
+ consumer.close();
+
+ //Wait longer than scan period.
+ Thread.sleep(20);
+
+ queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertNull(queueBinding);
+
+
+
+
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Test
+ public void testAutoDeleteOff() throws Exception {
+ ConnectionFactory fact = getCF();
+ Connection connection = fact.createConnection();
+ connection.start();
+
+ try {
+
+ Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+
+ String testQueueName = getName();
+
+ Queue queue = session.createQueue(testQueueName + "?auto-delete=false");
+ ActiveMQDestination activeMQDestination = (ActiveMQDestination) queue;
+
+ assertEquals(testQueueName, queue.getQueueName());
+ assertEquals(false, activeMQDestination.getQueueAttributes().getAutoDelete());
+
+ MessageProducer producer = session.createProducer(queue);
+ producer.send(session.createTextMessage("hello1"));
+ producer.send(session.createTextMessage("hello2"));
+
+ QueueBinding queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertFalse(queueBinding.getQueue().isAutoDelete());
+ assertEquals(2, queueBinding.getQueue().getMessageCount());
+
+ MessageConsumer consumer = session.createConsumer(queue);
+ Message message = consumer.receive(100);
+ assertNotNull(message);
+ message.acknowledge();
+
+ consumer.close();
+
+ queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertEquals(1, queueBinding.getQueue().getMessageCount());
+
+ consumer = session.createConsumer(queue);
+ message = consumer.receive(100);
+ assertNotNull(message);
+ message.acknowledge();
+
+ consumer.close();
+
+ //Wait longer than scan period.
+ Thread.sleep(20);
+
+ queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertNotNull(queueBinding);
+ assertEquals(0, queueBinding.getQueue().getMessageCount());
+
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Test
+ public void testAutoDeleteDelay() throws Exception {
+ ConnectionFactory fact = getCF();
+ Connection connection = fact.createConnection();
+ connection.start();
+
+ try {
+
+ Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+
+ String testQueueName = getName();
+
+ Queue queue = session.createQueue(testQueueName + "?auto-delete=true&auto-delete-delay=500");
+ ActiveMQDestination activeMQDestination = (ActiveMQDestination) queue;
+
+ assertEquals(testQueueName, queue.getQueueName());
+ assertEquals(Long.valueOf(500), activeMQDestination.getQueueAttributes().getAutoDeleteDelay());
+
+ MessageProducer producer = session.createProducer(queue);
+ producer.send(session.createTextMessage("hello1"));
+ producer.send(session.createTextMessage("hello2"));
+
+ QueueBinding queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertTrue(queueBinding.getQueue().isAutoDelete());
+ assertEquals(500, queueBinding.getQueue().getAutoDeleteDelay());
+ assertEquals(2, queueBinding.getQueue().getMessageCount());
+
+ MessageConsumer consumer = session.createConsumer(queue);
+ Message message = consumer.receive(100);
+ assertNotNull(message);
+ message.acknowledge();
+
+ consumer.close();
+
+ queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertEquals(1, queueBinding.getQueue().getMessageCount());
+
+ consumer = session.createConsumer(queue);
+ message = consumer.receive(100);
+ assertNotNull(message);
+ message.acknowledge();
+
+ consumer.close();
+
+ //Wait longer than scan period, but less than delay
+ Thread.sleep(50);
+
+ //Check the queue has not been removed.
+ queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertNotNull(queueBinding);
+
+ //Wait longer than auto delete delay
+ Thread.sleep(550);
+
+ queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertNull(queueBinding);
+
+
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Test
+ public void testAutoDeleteMessageCount() throws Exception {
+ ConnectionFactory fact = getCF();
+ Connection connection = fact.createConnection();
+ connection.start();
+
+ try {
+
+ Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+
+ String testQueueName = getName();
+
+ Queue queue = session.createQueue(testQueueName + "?auto-delete=true&auto-delete-message-count=1");
+ ActiveMQDestination activeMQDestination = (ActiveMQDestination) queue;
+
+ assertEquals(testQueueName, queue.getQueueName());
+ assertEquals(Long.valueOf(1), activeMQDestination.getQueueAttributes().getAutoDeleteMessageCount());
+
+ MessageProducer producer = session.createProducer(queue);
+ producer.send(session.createTextMessage("hello1"));
+ producer.send(session.createTextMessage("hello2"));
+
+ QueueBinding queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertTrue(queueBinding.getQueue().isAutoDelete());
+ assertEquals(1, queueBinding.getQueue().getAutoDeleteMessageCount());
+ assertEquals(2, queueBinding.getQueue().getMessageCount());
+
+ MessageConsumer consumer = session.createConsumer(queue);
+ Message message = consumer.receive(100);
+ assertNotNull(message);
+ message.acknowledge();
+
+ consumer.close();
+
+ //Wait longer than scan period
+ Thread.sleep(20);
+
+ queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertNull(queueBinding);
+
+
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Test
+ public void testAutoDeleteMessageCountDisabled() throws Exception {
+ ConnectionFactory fact = getCF();
+ Connection connection = fact.createConnection();
+ connection.start();
+
+ try {
+
+ Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+
+ String testQueueName = getName();
+
+ Queue queue = session.createQueue(testQueueName + "?auto-delete=true&auto-delete-message-count=-1");
+ ActiveMQDestination activeMQDestination = (ActiveMQDestination) queue;
+
+ assertEquals(testQueueName, queue.getQueueName());
+ assertEquals(Long.valueOf(-1), activeMQDestination.getQueueAttributes().getAutoDeleteMessageCount());
+
+ MessageProducer producer = session.createProducer(queue);
+ for (int i = 0; i < 100; i++) {
+ producer.send(session.createTextMessage("hello" + i));
+ }
+
+ QueueBinding queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertEquals(100, queueBinding.getQueue().getMessageCount());
+ assertTrue(queueBinding.getQueue().isAutoDelete());
+ assertEquals(-1, queueBinding.getQueue().getAutoDeleteMessageCount());
+
+ MessageConsumer consumer = session.createConsumer(queue);
+ Message message = consumer.receive(100);
+ assertNotNull(message);
+ message.acknowledge();
+
+ consumer.close();
+
+ //Wait longer than scan period
+ Thread.sleep(20);
+
+ queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(testQueueName));
+ assertNull(queueBinding);
+
+
+ } finally {
+ connection.close();
+ }
+ }
+
+
+}
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 a3d7786..e77b96d 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
@@ -205,7 +205,7 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
}
@Override
- public String createQueue(String address, String routingType, String name, String filter, boolean durable, int maxConsumers, boolean purgeOnNoConsumers, boolean exclusive, boolean groupRebalance, int groupBuckets, boolean lastValue, String lastValueKey, boolean nonDestructive, int consumersBeforeDispatch, long delayBeforeDispatch, boolean autoCreateAddress) throws Exception {
+ public String createQueue(String address, String routingType, String name, String filter, boolean durable, int maxConsumers, boolean purgeOnNoConsumers, boolean exclusive, boolean groupRebalance, int groupBuckets, boolean lastValue, String lastValueKey, boolean nonDestructive, int consumersBeforeDispatch, long delayBeforeDispatch, boolean autoDelete, long autoDeleteDelay, long autoDeleteMessageCount, boolean autoCreateAddress) throws Exception {
return (String) proxy.invokeOperation("createQueue", address, routingType, name, filter, durable, maxConsumers, purgeOnNoConsumers, exclusive, groupRebalance, groupBuckets, lastValue, lastValueKey, nonDestructive, consumersBeforeDispatch, delayBeforeDispatch, autoCreateAddress);
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/QueueConfigRestartTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/QueueConfigRestartTest.java
index 421013a..fffa66d 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/QueueConfigRestartTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/QueueConfigRestartTest.java
@@ -115,7 +115,7 @@ public class QueueConfigRestartTest extends ActiveMQTestBase {
SimpleString address = new SimpleString("test.address");
SimpleString queue = new SimpleString("test.queue");
- server.createQueue(address, RoutingType.MULTICAST, queue, null, null, true, false, false, false,false, 10, true, true, false, -1, true, null, false, consumersBeforeDispatch, -1, true);
+ server.createQueue(address, RoutingType.MULTICAST, queue, null, null, true, false, false, false,false, 10, true, true, false, -1, true, null, false, consumersBeforeDispatch, -1, false, 0, 0, true);
QueueBinding queueBinding1 = (QueueBinding)server.getPostOffice().getBinding(queue);
Assert.assertEquals(consumersBeforeDispatch, queueBinding1.getQueue().getConsumersBeforeDispatch());
@@ -138,7 +138,7 @@ public class QueueConfigRestartTest extends ActiveMQTestBase {
SimpleString address = new SimpleString("test.address");
SimpleString queue = new SimpleString("test.queue");
- server.createQueue(address, RoutingType.MULTICAST, queue, null, null, true, false, false, false,false, 10, true, true, false, -1, true, null, false,0, delayBeforeDispatch, true);
+ server.createQueue(address, RoutingType.MULTICAST, queue, null, null, true, false, false, false,false, 10, true, true, false, -1, true, null, false,0, delayBeforeDispatch, false, 0, 0, true);
QueueBinding queueBinding1 = (QueueBinding)server.getPostOffice().getBinding(queue);
Assert.assertEquals(delayBeforeDispatch, queueBinding1.getQueue().getDelayBeforeDispatch());
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
index e81728c..0aa631b 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
@@ -627,6 +627,24 @@ public class FakeQueue extends CriticalComponentImpl implements Queue {
}
@Override
+ public boolean isAutoDelete() {
+ // no-op
+ return false;
+ }
+
+ @Override
+ public long getAutoDeleteDelay() {
+ // no-op
+ return -1;
+ }
+
+ @Override
+ public long getAutoDeleteMessageCount() {
+ // no-op
+ return -1;
+ }
+
+ @Override
public boolean isTemporary() {
// no-op
return false;