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 2018/10/18 00:53:23 UTC
[4/5] activemq-artemis git commit: ARTEMIS-2117 Add custom LVQ Key
and Non Destructive Queue into Broker
ARTEMIS-2117 Add custom LVQ Key and Non Destructive Queue into Broker
Implement custom LVQ Key and Non-Destructive in broker - protocol agnostic
Make feature configurable via broker.xml, core apis and activemqservercontrol
Add last-value-key test cases
Add non-destructive with lvq test cases
Add non-destructive with expiry-delay test cases
Update documents
Add new methods to support create, update with new attributes
Refactor to pass through queue-attributes in client side methods to reduce further method changes for adding new attributes in future and avoid methods with endless parameters. (note: in future this should prob be done server side too)
Update existing test cases and fake impls for new methods/attributes
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/547b2aa5
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/547b2aa5
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/547b2aa5
Branch: refs/heads/master
Commit: 547b2aa592138b3d53d295f53eef94a499073aab
Parents: 44fa690
Author: Michael André Pearce <mi...@me.com>
Authored: Sun Oct 14 22:02:13 2018 +0100
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Oct 17 20:53:13 2018 -0400
----------------------------------------------------------------------
.../artemis/api/core/QueueAttributes.java | 106 ++++-
.../config/ActiveMQDefaultConfiguration.java | 14 +-
.../artemis/api/core/client/ClientSession.java | 49 ++
.../core/management/ActiveMQServerControl.java | 4 +
.../core/client/impl/AddressQueryImpl.java | 38 +-
.../core/client/impl/ClientSessionImpl.java | 176 ++++---
.../core/client/impl/QueueQueryImpl.java | 57 +++
.../core/impl/ActiveMQSessionContext.java | 90 +++-
.../impl/wireformat/CreateQueueMessage_V2.java | 115 ++++-
.../wireformat/CreateSharedQueueMessage_V2.java | 84 +++-
.../SessionBindingQueryResponseMessage_V4.java | 76 ++-
.../SessionQueueQueryResponseMessage_V3.java | 96 +++-
.../artemis/core/server/QueueQueryResult.java | 36 ++
.../spi/core/remoting/SessionContext.java | 20 +
.../jms/client/ActiveMQMessageProducer.java | 40 +-
.../artemis/jms/client/ActiveMQSession.java | 132 +++---
.../client/HornetQClientSessionContext.java | 2 +-
.../core/protocol/openwire/amq/AMQConsumer.java | 4 +-
.../core/config/CoreQueueConfiguration.java | 38 ++
.../deployers/impl/FileConfigurationParser.java | 16 +-
.../impl/ActiveMQServerControlImpl.java | 33 +-
.../core/paging/cursor/PagedReferenceImpl.java | 10 +
.../core/persistence/QueueBindingInfo.java | 8 +
.../journal/AbstractJournalStorageManager.java | 2 +-
.../codec/PersistentQueueBindingEncoding.java | 46 ++
.../artemis/core/postoffice/PostOffice.java | 1 +
.../core/postoffice/impl/PostOfficeImpl.java | 5 +
.../core/ServerSessionPacketHandler.java | 9 +-
.../artemis/core/server/ActiveMQServer.java | 13 +-
.../artemis/core/server/BindingQueryResult.java | 38 +-
.../artemis/core/server/MessageReference.java | 3 +
.../activemq/artemis/core/server/Queue.java | 10 +-
.../artemis/core/server/QueueConfig.java | 38 +-
.../artemis/core/server/ServerSession.java | 30 ++
.../artemis/core/server/impl/AckReason.java | 2 +-
.../core/server/impl/ActiveMQServerImpl.java | 72 ++-
.../core/server/impl/LastValueQueue.java | 77 ++-
.../core/server/impl/MessageReferenceImpl.java | 10 +
.../server/impl/PostOfficeJournalLoader.java | 2 +
.../core/server/impl/QueueFactoryImpl.java | 31 +-
.../artemis/core/server/impl/QueueImpl.java | 272 ++++++-----
.../core/server/impl/ServerSessionImpl.java | 68 ++-
.../core/settings/impl/AddressSettings.java | 62 ++-
.../resources/schema/artemis-configuration.xsd | 20 +
.../impl/ScheduledDeliveryHandlerTest.java | 15 +
.../test/resources/artemis-configuration.xsd | 20 +
docs/user-manual/en/last-value-queues.md | 120 ++++-
docs/user-manual/en/message-expiry.md | 21 +
.../integration/DuplicateDetectionTest.java | 2 +-
.../integration/amqp/JMSClientTestSupport.java | 4 +
.../tests/integration/amqp/JMSLVQTest.java | 129 +++--
.../integration/amqp/JMSNonDestructiveTest.java | 467 +++++++++++++++++++
.../jms/client/ConsumerDelayDispatchTest.java | 4 +-
.../tests/integration/jms/client/LVQTest.java | 94 ++++
.../ActiveMQServerControlUsingCoreTest.java | 7 +-
.../persistence/QueueConfigRestartTest.java | 4 +-
.../jms/tests/message/MessageHeaderTest.java | 18 +-
.../unit/core/postoffice/impl/FakeQueue.java | 15 +
.../core/server/impl/fakes/FakePostOffice.java | 2 +-
59 files changed, 2522 insertions(+), 455 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/QueueAttributes.java
----------------------------------------------------------------------
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 6a43b39..12c5f86 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
@@ -21,59 +21,155 @@ import java.io.Serializable;
public class QueueAttributes implements Serializable {
+ public static final String ROUTING_TYPE = "routing-type";
+ public static final String FILTER_STRING = "filter-string";
+ public static final String DURABLE = "durable";
public static final String MAX_CONSUMERS = "max-consumers";
public static final String EXCLUSIVE = "exclusive";
public static final String LAST_VALUE = "last-value";
+ public static final String LAST_VALUE_KEY = "last-value-key";
+ public static final String NON_DESTRUCTIVE = "non-destructive";
public static final String PURGE_ON_NO_CONSUMERS = "purge-on-no-consumers";
+ public static final String CONSUMERS_BEFORE_DISPATCH = "consumers-before-dispatch";
+ public static final String DELAY_BEFORE_DISPATCH = "delay-before-dispatch";
+ private RoutingType routingType;
+ private SimpleString filterString;
+ private Boolean durable;
private Integer maxConsumers;
private Boolean exclusive;
private Boolean lastValue;
+ private SimpleString lastValueKey;
+ private Boolean nonDestructive;
private Boolean purgeOnNoConsumers;
+ private Integer consumersBeforeDispatch;
+ private Long delayBeforeDispatch;
public void set(String key, String value) {
if (key != null && value != null) {
- if (key.equals(MAX_CONSUMERS)) {
+ if (key.equals(ROUTING_TYPE)) {
+ setRoutingType(RoutingType.valueOf(value.toUpperCase()));
+ } else if (key.equals(FILTER_STRING)) {
+ setFilterString(SimpleString.toSimpleString(value));
+ } else if (key.equals(DURABLE)) {
+ setDurable(Boolean.valueOf(value));
+ } else if (key.equals(MAX_CONSUMERS)) {
setMaxConsumers(Integer.valueOf(value));
} else if (key.equals(EXCLUSIVE)) {
setExclusive(Boolean.valueOf(value));
} else if (key.equals(LAST_VALUE)) {
setLastValue(Boolean.valueOf(value));
+ } else if (key.equals(LAST_VALUE_KEY)) {
+ setLastValueKey(SimpleString.toSimpleString(value));
+ } else if (key.equals(NON_DESTRUCTIVE)) {
+ setNonDestructive(Boolean.valueOf(value));
} else if (key.equals(PURGE_ON_NO_CONSUMERS)) {
setPurgeOnNoConsumers(Boolean.valueOf(value));
+ } else if (key.equals(CONSUMERS_BEFORE_DISPATCH)) {
+ setConsumersBeforeDispatch(Integer.valueOf(value));
+ } else if (key.equals(DELAY_BEFORE_DISPATCH)) {
+ setDelayBeforeDispatch(Long.valueOf(value));
}
}
}
+ public RoutingType getRoutingType() {
+ return routingType;
+ }
+
+ public QueueAttributes setRoutingType(RoutingType routingType) {
+ this.routingType = routingType;
+ return this;
+ }
+
+ public SimpleString getFilterString() {
+ return filterString;
+ }
+
+ public QueueAttributes setFilterString(SimpleString filterString) {
+ this.filterString = filterString;
+ return this;
+ }
+
+ public Boolean getDurable() {
+ return durable;
+ }
+
+ public QueueAttributes setDurable(Boolean durable) {
+ this.durable = durable;
+ return this;
+ }
+
public Integer getMaxConsumers() {
return maxConsumers;
}
- public void setMaxConsumers(Integer maxConsumers) {
+ public QueueAttributes setMaxConsumers(Integer maxConsumers) {
this.maxConsumers = maxConsumers;
+ return this;
}
public Boolean getExclusive() {
return exclusive;
}
- public void setExclusive(Boolean exclusive) {
+ public QueueAttributes setExclusive(Boolean exclusive) {
this.exclusive = exclusive;
+ return this;
}
public Boolean getLastValue() {
return lastValue;
}
- public void setLastValue(Boolean lastValue) {
+ public QueueAttributes setLastValue(Boolean lastValue) {
this.lastValue = lastValue;
+ return this;
+ }
+
+ public SimpleString getLastValueKey() {
+ return lastValueKey;
+ }
+
+ public QueueAttributes setLastValueKey(SimpleString lastValueKey) {
+ this.lastValueKey = lastValueKey;
+ return this;
+ }
+
+ public Boolean getNonDestructive() {
+ return nonDestructive;
+ }
+
+ public QueueAttributes setNonDestructive(Boolean nonDestructive) {
+ this.nonDestructive = nonDestructive;
+ return this;
}
public Boolean getPurgeOnNoConsumers() {
return purgeOnNoConsumers;
}
- public void setPurgeOnNoConsumers(Boolean purgeOnNoConsumers) {
+ public QueueAttributes setPurgeOnNoConsumers(Boolean purgeOnNoConsumers) {
this.purgeOnNoConsumers = purgeOnNoConsumers;
+ return this;
+ }
+
+ public Integer getConsumersBeforeDispatch() {
+ return consumersBeforeDispatch;
}
+
+ public QueueAttributes setConsumersBeforeDispatch(Integer consumersBeforeDispatch) {
+ this.consumersBeforeDispatch = consumersBeforeDispatch;
+ return this;
+ }
+
+ public Long getDelayBeforeDispatch() {
+ return delayBeforeDispatch;
+ }
+
+ public QueueAttributes setDelayBeforeDispatch(Long delayBeforeDispatch) {
+ this.delayBeforeDispatch = delayBeforeDispatch;
+ return this;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
index bd7ce51..e05a88e 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
@@ -471,6 +471,10 @@ public final class ActiveMQDefaultConfiguration {
public static final boolean DEFAULT_LAST_VALUE = false;
+ public static final SimpleString DEFAULT_LAST_VALUE_KEY = null;
+
+ public static final boolean DEFAULT_NON_DESTRUCTIVE = false;
+
public static final boolean DEFAULT_PURGE_ON_NO_CONSUMERS = false;
public static final int DEFAULT_CONSUMERS_BEFORE_DISPATCH = 0;
@@ -1301,7 +1305,15 @@ public final class ActiveMQDefaultConfiguration {
}
public static boolean getDefaultLastValue() {
- return DEFAULT_EXCLUSIVE;
+ return DEFAULT_LAST_VALUE;
+ }
+
+ public static SimpleString getDefaultLastValueKey() {
+ return DEFAULT_LAST_VALUE_KEY;
+ }
+
+ public static boolean getDefaultNonDestructive() {
+ return DEFAULT_NON_DESTRUCTIVE;
}
public static boolean getDefaultPurgeOnNoConsumers() {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/ClientSession.java
----------------------------------------------------------------------
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 414def3..beb91e9 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
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.Set;
import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.QueueAttributes;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.RoutingType;
@@ -79,6 +80,14 @@ public interface ClientSession extends XAResource, AutoCloseable {
Boolean isDefaultLastValueQueue();
Boolean isDefaultExclusive();
+
+ SimpleString getDefaultLastValueKey();
+
+ Boolean isDefaultNonDestructive();
+
+ Integer getDefaultConsumersBeforeDispatch();
+
+ Long getDefaultDelayBeforeDispatch();
}
/**
@@ -148,6 +157,14 @@ public interface ClientSession extends XAResource, AutoCloseable {
Boolean isLastValue();
+ SimpleString getLastValueKey();
+
+ Boolean isNonDestructive();
+
+ Integer getConsumersBeforeDispatch();
+
+ Long getDelayBeforeDispatch();
+
Integer getDefaultConsumerWindowSize();
}
@@ -482,6 +499,17 @@ public interface ClientSession extends XAResource, AutoCloseable {
boolean durable, Integer maxConsumers, Boolean purgeOnNoConsumers, Boolean exclusive, Boolean lastValue) throws ActiveMQException;
/**
+ * Creates Shared queue. A queue that will exist as long as there are consumers or is durable.
+ *
+ * @param address the queue will be bound to this address
+ * @param queueName the name of the queue
+ * @param queueAttributes attributes for the queue
+ * @throws ActiveMQException in an exception occurs while creating the queue
+ */
+ void createSharedQueue(SimpleString address, SimpleString queueName, QueueAttributes queueAttributes) throws ActiveMQException;
+
+
+ /**
* Creates a <em>non-temporary</em> queue.
*
* @param address the queue will be bound to this address
@@ -586,6 +614,17 @@ public interface ClientSession extends XAResource, AutoCloseable {
boolean durable, boolean autoCreated, int maxConsumers, boolean purgeOnNoConsumers, Boolean exclusive, Boolean lastValue) throws ActiveMQException;
/**
+ * Creates a <em>non-temporary</em> queue.
+ *
+ * @param address the queue will be bound to this address
+ * @param queueName the name of the queue
+ * @param autoCreated whether to mark this queue as autoCreated or not
+ * @param queueAttributes attributes for the queue
+ * @throws ActiveMQException
+ */
+ void createQueue(SimpleString address, SimpleString queueName, boolean autoCreated, QueueAttributes queueAttributes) throws ActiveMQException;
+
+ /**
* Creates a <em>non-temporary</em>queue.
*
* @param address the queue will be bound to this address
@@ -672,6 +711,16 @@ public interface ClientSession extends XAResource, AutoCloseable {
* Creates a <em>temporary</em> queue with a filter.
*
* @param address the queue will be bound to this address
+ * @param queueName the name of the queue
+ * @param queueAttributes attributes for the queue
+ * @throws ActiveMQException in an exception occurs while creating the queue
+ */
+ void createTemporaryQueue(SimpleString address, SimpleString queueName, QueueAttributes queueAttributes) throws ActiveMQException;
+
+ /**
+ * Creates a <em>temporary</em> queue with a filter.
+ *
+ * @param address the queue will be bound to this address
* @param routingType the routing type for this queue, MULTICAST or ANYCAST
* @param queueName the name of the queue
* @param filter only messages which match this filter will be put in the queue
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java
----------------------------------------------------------------------
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 5719fb6..62a422f 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
@@ -599,6 +599,8 @@ public interface ActiveMQServerControl {
@Parameter(name = "purgeOnNoConsumers", desc = "Delete this queue when the last consumer disconnects") boolean purgeOnNoConsumers,
@Parameter(name = "exclusive", desc = "If the queue should route exclusively to one consumer") boolean exclusive,
@Parameter(name = "lastValue", desc = "Use last-value semantics") boolean lastValue,
+ @Parameter(name = "lastValueKey", desc = "Use the specified property key for the last value") String lastValueKey,
+ @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 = "autoCreateAddress", desc = "Create an address with default values should a matching address not be found") boolean autoCreateAddress) throws Exception;
@@ -696,6 +698,7 @@ public interface ActiveMQServerControl {
* @param maxConsumers the maximum number of consumers allowed on this queue at any one time
* @param purgeOnNoConsumers delete this queue when the last consumer disconnects
* @param exclusive if the queue should route exclusively to one consumer
+ * @param nonDestructive If the queue is non-destructive
* @param consumersBeforeDispatch number of consumers needed before dispatch can start
* @param delayBeforeDispatch delay to wait before dispatching if number of consumers before dispatch is not met
* @param user the user associated with this queue
@@ -709,6 +712,7 @@ public interface ActiveMQServerControl {
@Parameter(name = "maxConsumers", desc = "The maximum number of consumers allowed on this queue at any one time") Integer maxConsumers,
@Parameter(name = "purgeOnNoConsumers", desc = "Delete this queue when the last consumer disconnects") Boolean purgeOnNoConsumers,
@Parameter(name = "exclusive", desc = "If the queue should route exclusively to one consumer") Boolean exclusive,
+ @Parameter(name = "nonDestructive", desc = "If the queue is non-destructive") Boolean nonDestructive,
@Parameter(name = "consumersBeforeDispatch", desc = "Number of consumers needed before dispatch can start") Integer consumersBeforeDispatch,
@Parameter(name = "delayBeforeDispatch", desc = "Delay to wait before dispatching if number of consumers before dispatch is not met") Long delayBeforeDispatch,
@Parameter(name = "user", desc = "The user associated with this queue") String user) throws Exception;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/AddressQueryImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/AddressQueryImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/AddressQueryImpl.java
index cb91919..8c543de 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/AddressQueryImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/AddressQueryImpl.java
@@ -40,6 +40,14 @@ public class AddressQueryImpl implements ClientSession.AddressQuery {
private final Boolean defaultLastValue;
+ private final SimpleString defaultLastValueKey;
+
+ private final Boolean defaultNonDestructive;
+
+ private final Integer defaultConsumersBeforeDispatch;
+
+ private final Long defaultDelayBeforeDispatch;
+
public AddressQueryImpl(final boolean exists,
final List<SimpleString> queueNames,
final boolean autoCreateQueues,
@@ -47,7 +55,11 @@ public class AddressQueryImpl implements ClientSession.AddressQuery {
final boolean defaultPurgeOnNoConsumers,
final int defaultMaxConsumers,
final Boolean defaultExclusive,
- final Boolean defaultLastValue) {
+ final Boolean defaultLastValue,
+ final SimpleString defaultLastValueKey,
+ final Boolean defaultNonDestructive,
+ final Integer defaultConsumersBeforeDispatch,
+ final Long defaultDelayBeforeDispatch) {
this.exists = exists;
this.queueNames = new ArrayList<>(queueNames);
this.autoCreateQueues = autoCreateQueues;
@@ -56,6 +68,10 @@ public class AddressQueryImpl implements ClientSession.AddressQuery {
this.defaultMaxConsumers = defaultMaxConsumers;
this.defaultExclusive = defaultExclusive;
this.defaultLastValue = defaultLastValue;
+ this.defaultLastValueKey = defaultLastValueKey;
+ this.defaultNonDestructive = defaultNonDestructive;
+ this.defaultConsumersBeforeDispatch = defaultConsumersBeforeDispatch;
+ this.defaultDelayBeforeDispatch = defaultDelayBeforeDispatch;
}
@Override
@@ -97,4 +113,24 @@ public class AddressQueryImpl implements ClientSession.AddressQuery {
public Boolean isDefaultExclusive() {
return defaultExclusive;
}
+
+ @Override
+ public SimpleString getDefaultLastValueKey() {
+ return defaultLastValueKey;
+ }
+
+ @Override
+ public Boolean isDefaultNonDestructive() {
+ return defaultNonDestructive;
+ }
+
+ @Override
+ public Integer getDefaultConsumersBeforeDispatch() {
+ return defaultConsumersBeforeDispatch;
+ }
+
+ @Override
+ public Long getDefaultDelayBeforeDispatch() {
+ return defaultDelayBeforeDispatch;
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java
index 2edf629..5efaed6 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java
@@ -34,6 +34,8 @@ import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.QueueAttributes;
+import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ClientConsumer;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
@@ -46,7 +48,6 @@ import org.apache.activemq.artemis.core.client.ActiveMQClientLogger;
import org.apache.activemq.artemis.core.client.ActiveMQClientMessageBundle;
import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
import org.apache.activemq.artemis.core.remoting.FailureListener;
-import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
import org.apache.activemq.artemis.spi.core.remoting.ConsumerContext;
import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
@@ -336,7 +337,7 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
@Override
public void createTemporaryQueue(final SimpleString address, final SimpleString queueName) throws ActiveMQException {
- createTemporaryQueue(address, queueName, null);
+ createTemporaryQueue(address, queueName, (SimpleString) null);
}
@Override
@@ -370,13 +371,15 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
final boolean durable,
final boolean autoCreated) throws ActiveMQException {
internalCreateQueue(address,
- queueName, routingType,
- filterString,
- durable,
+ queueName,
false,
- ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(),
- ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(),
- autoCreated, null, null);
+ autoCreated,
+ new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(filterString)
+ .setDurable(durable)
+ .setPurgeOnNoConsumers(ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers())
+ .setMaxConsumers(ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers()));
}
@Override
@@ -394,28 +397,41 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
public void createQueue(final SimpleString address, final RoutingType routingType, final SimpleString queueName, final SimpleString filterString,
final boolean durable, final boolean autoCreated, final int maxConsumers, final boolean purgeOnNoConsumers) throws ActiveMQException {
internalCreateQueue(address,
- queueName, routingType,
- filterString,
- durable,
+ queueName,
false,
- maxConsumers,
- purgeOnNoConsumers,
- autoCreated, null, null);
+ autoCreated,
+ new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(filterString)
+ .setDurable(durable)
+ .setMaxConsumers(maxConsumers)
+ .setPurgeOnNoConsumers(purgeOnNoConsumers));
}
@Override
public void createQueue(final SimpleString address, final RoutingType routingType, final SimpleString queueName, final SimpleString filterString,
final boolean durable, final boolean autoCreated, final int maxConsumers, final boolean purgeOnNoConsumers, final Boolean exclusive, final Boolean lastValue) throws ActiveMQException {
internalCreateQueue(address,
- queueName, routingType,
- filterString,
- durable,
+ queueName,
false,
- maxConsumers,
- purgeOnNoConsumers,
autoCreated,
- exclusive,
- lastValue);
+ new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(filterString)
+ .setDurable(durable)
+ .setMaxConsumers(maxConsumers)
+ .setPurgeOnNoConsumers(purgeOnNoConsumers)
+ .setExclusive(exclusive)
+ .setLastValue(lastValue));
+ }
+
+ @Override
+ public void createQueue(final SimpleString address, final SimpleString queueName, final boolean autoCreated, final QueueAttributes queueAttributes) throws ActiveMQException {
+ internalCreateQueue(address,
+ queueName,
+ false,
+ autoCreated,
+ queueAttributes);
}
@Override
@@ -470,13 +486,28 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
final Boolean exclusive,
final Boolean lastValue) throws ActiveMQException {
internalCreateQueue(address,
- queueName, routingType,
- filter,
- false,
+ queueName,
true,
- maxConsumers,
- purgeOnNoConsumers,
- false, exclusive, lastValue);
+ false,
+ new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(filter)
+ .setDurable(false)
+ .setPurgeOnNoConsumers(purgeOnNoConsumers)
+ .setMaxConsumers(maxConsumers)
+ .setExclusive(exclusive)
+ .setLastValue(lastValue));
+ }
+
+ @Override
+ public void createTemporaryQueue(final SimpleString address,
+ final SimpleString queueName,
+ final QueueAttributes queueAttributes) throws ActiveMQException {
+ internalCreateQueue(address,
+ queueName,
+ true,
+ false,
+ queueAttributes);
}
@Override
@@ -504,13 +535,15 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
@Override
public void createQueue(SimpleString address, RoutingType routingType, SimpleString queueName, boolean durable) throws ActiveMQException {
internalCreateQueue(address,
- queueName, routingType,
- null,
- durable,
+ queueName,
+ false,
false,
- ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(),
- ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(),
- false, null, null);
+ new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(null)
+ .setDurable(durable)
+ .setPurgeOnNoConsumers(ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers())
+ .setMaxConsumers(ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers()));
}
/**
@@ -564,11 +597,32 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
@Override
public void createSharedQueue(SimpleString address, RoutingType routingType, SimpleString queueName, SimpleString filter,
boolean durable, Integer maxConsumers, Boolean purgeOnNoConsumers, Boolean exclusive, Boolean lastValue) throws ActiveMQException {
+ QueueAttributes queueAttributes = new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(filter)
+ .setDurable(durable)
+ .setPurgeOnNoConsumers(ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers())
+ .setMaxConsumers(ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers())
+ .setExclusive(exclusive)
+ .setLastValue(lastValue);
+ createSharedQueue(address, queueName, queueAttributes);
+ }
+
+ /**
+ * Creates Shared queue. A queue that will exist as long as there are consumers or is durable.
+ *
+ * @param address the queue will be bound to this address
+ * @param queueName the name of the queue
+ * @param queueAttributes attributes for the queue
+ * @throws ActiveMQException in an exception occurs while creating the queue
+ */
+ @Override
+ public void createSharedQueue(SimpleString address, SimpleString queueName, QueueAttributes queueAttributes) throws ActiveMQException {
checkClosed();
startCall();
try {
- sessionContext.createSharedQueue(address, queueName, routingType, filter, durable, maxConsumers, purgeOnNoConsumers, exclusive, lastValue);
+ sessionContext.createSharedQueue(address, queueName, queueAttributes);
} finally {
endCall();
}
@@ -599,13 +653,15 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
@Override
public void createQueue(String address, RoutingType routingType, String queueName) throws ActiveMQException {
internalCreateQueue(SimpleString.toSimpleString(address),
- SimpleString.toSimpleString(queueName), routingType,
- null,
+ SimpleString.toSimpleString(queueName),
false,
false,
- ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(),
- ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(),
- false, null, null);
+ new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(null)
+ .setDurable(false)
+ .setPurgeOnNoConsumers(ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers())
+ .setMaxConsumers(ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers()));
}
/**
@@ -620,13 +676,14 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
public void createQueue(SimpleString address, RoutingType routingType, SimpleString queueName) throws ActiveMQException {
internalCreateQueue(address,
queueName,
- routingType,
- null,
false,
false,
- ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(),
- ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(),
- false, null, null);
+ new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(null)
+ .setDurable(false)
+ .setPurgeOnNoConsumers(ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers())
+ .setMaxConsumers(ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers()));
}
/**
@@ -644,13 +701,14 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
boolean durable) throws ActiveMQException {
internalCreateQueue(address,
queueName,
- routingType,
- filter,
- durable,
false,
- ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(),
- ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(),
- false, null, null);
+ false,
+ new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(filter)
+ .setDurable(durable)
+ .setPurgeOnNoConsumers(ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers())
+ .setMaxConsumers(ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers()));
}
/**
@@ -1907,34 +1965,22 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
private void internalCreateQueue(final SimpleString address,
final SimpleString queueName,
- final RoutingType routingType,
- final SimpleString filterString,
- final boolean durable,
final boolean temp,
- final int maxConsumers,
- final boolean purgeOnNoConsumers,
final boolean autoCreated,
- final Boolean exclusive,
- final Boolean lastValue) throws ActiveMQException {
+ final QueueAttributes queueAttributes) throws ActiveMQException {
checkClosed();
- if (durable && temp) {
+ if (queueAttributes.getDurable() && temp) {
throw ActiveMQClientMessageBundle.BUNDLE.queueMisConfigured();
}
startCall();
try {
sessionContext.createQueue(address,
- routingType,
queueName,
- filterString,
- durable,
temp,
- maxConsumers,
- purgeOnNoConsumers,
autoCreated,
- exclusive,
- lastValue);
+ queueAttributes);
} finally {
endCall();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/QueueQueryImpl.java
----------------------------------------------------------------------
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 d377d18..550728b 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
@@ -52,6 +52,15 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
private final Boolean lastValue;
+ private final SimpleString lastValueKey;
+
+ private final Boolean nonDestructive;
+
+ private final Integer consumersBeforeDispatch;
+
+ private final Long delayBeforeDispatch;
+
+
private final Integer defaultConsumerWindowSize;
public QueueQueryImpl(final boolean durable,
@@ -92,6 +101,26 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
final RoutingType routingType) {
this(durable, temporary, consumerCount, messageCount, filterString, address, name, exists, autoCreateQueues, maxConsumers, autoCreated, purgeOnNoConsumers, routingType, null, null, null);
}
+
+ public QueueQueryImpl(final boolean durable,
+ final boolean temporary,
+ final int consumerCount,
+ final long messageCount,
+ final SimpleString filterString,
+ final SimpleString address,
+ final SimpleString name,
+ final boolean exists,
+ final boolean autoCreateQueues,
+ final int maxConsumers,
+ final boolean autoCreated,
+ final boolean purgeOnNoConsumers,
+ final RoutingType routingType,
+ final Boolean exclusive,
+ final Boolean lastValue,
+ final Integer defaultConsumerWindowSize) {
+ this(durable, temporary, consumerCount, messageCount, filterString, address, name, exists, autoCreateQueues, maxConsumers, autoCreated, purgeOnNoConsumers, routingType, exclusive, lastValue, null, null, null, null, defaultConsumerWindowSize);
+ }
+
public QueueQueryImpl(final boolean durable,
final boolean temporary,
final int consumerCount,
@@ -107,6 +136,10 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
final RoutingType routingType,
final Boolean exclusive,
final Boolean lastValue,
+ final SimpleString lastValueKey,
+ final Boolean nonDestructive,
+ final Integer consumersBeforeDispatch,
+ final Long delayBeforeDispatch,
final Integer defaultConsumerWindowSize) {
this.durable = durable;
this.temporary = temporary;
@@ -123,6 +156,10 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
this.routingType = routingType;
this.exclusive = exclusive;
this.lastValue = lastValue;
+ this.lastValueKey = lastValueKey;
+ this.nonDestructive = nonDestructive;
+ this.consumersBeforeDispatch = consumersBeforeDispatch;
+ this.delayBeforeDispatch = delayBeforeDispatch;
this.defaultConsumerWindowSize = defaultConsumerWindowSize;
}
@@ -202,6 +239,26 @@ public class QueueQueryImpl implements ClientSession.QueueQuery {
}
@Override
+ public SimpleString getLastValueKey() {
+ return lastValueKey;
+ }
+
+ @Override
+ public Boolean isNonDestructive() {
+ return nonDestructive;
+ }
+
+ @Override
+ public Integer getConsumersBeforeDispatch() {
+ return consumersBeforeDispatch;
+ }
+
+ @Override
+ public Long getDelayBeforeDispatch() {
+ return delayBeforeDispatch;
+ }
+
+ @Override
public Integer getDefaultConsumerWindowSize() {
return defaultConsumerWindowSize;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
----------------------------------------------------------------------
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 7306072..3307b4d 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
@@ -16,6 +16,14 @@
*/
package org.apache.activemq.artemis.core.protocol.core.impl;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.QueueAttributes;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.DISCONNECT_CONSUMER;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.EXCEPTION;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_RECEIVE_CONTINUATION;
@@ -37,13 +45,6 @@ import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
-import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
-import org.apache.activemq.artemis.api.core.ActiveMQException;
-import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
-import org.apache.activemq.artemis.api.core.ICoreMessage;
-import org.apache.activemq.artemis.api.core.Message;
-import org.apache.activemq.artemis.api.core.RoutingType;
-import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
import org.apache.activemq.artemis.api.core.client.ClientConsumer;
import org.apache.activemq.artemis.api.core.client.ClientSession;
@@ -291,12 +292,40 @@ public class ActiveMQSessionContext extends SessionContext {
Boolean purgeOnNoConsumers,
Boolean exclusive,
Boolean lastValue) throws ActiveMQException {
- sessionChannel.sendBlocking(new CreateSharedQueueMessage_V2(address, queueName, routingType, filterString, durable, maxConsumers, purgeOnNoConsumers, exclusive, lastValue, true), PacketImpl.NULL_RESPONSE);
+ QueueAttributes queueAttributes = new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(filterString)
+ .setDurable(durable)
+ .setMaxConsumers(maxConsumers)
+ .setPurgeOnNoConsumers(purgeOnNoConsumers)
+ .setExclusive(exclusive)
+ .setLastValue(lastValue);
+ createSharedQueue(address, queueName, queueAttributes);
}
@Override
public void createSharedQueue(SimpleString address,
SimpleString queueName,
+ QueueAttributes queueAttributes) throws ActiveMQException {
+ sessionChannel.sendBlocking(new CreateSharedQueueMessage_V2(address, queueName,
+ queueAttributes.getRoutingType(),
+ queueAttributes.getFilterString(),
+ queueAttributes.getDurable(),
+ queueAttributes.getMaxConsumers(),
+ queueAttributes.getPurgeOnNoConsumers(),
+ queueAttributes.getExclusive(),
+ queueAttributes.getLastValue(),
+ queueAttributes.getLastValueKey(),
+ queueAttributes.getNonDestructive(),
+ queueAttributes.getConsumersBeforeDispatch(),
+ queueAttributes.getDelayBeforeDispatch(),
+ true), PacketImpl.NULL_RESPONSE);
+ }
+
+
+ @Override
+ public void createSharedQueue(SimpleString address,
+ SimpleString queueName,
RoutingType routingType,
SimpleString filterString,
boolean durable) throws ActiveMQException {
@@ -376,19 +405,19 @@ public class ActiveMQSessionContext extends SessionContext {
if (sessionChannel.supports(PacketImpl.SESS_BINDINGQUERY_RESP_V4, getServerVersion())) {
Packet packet = sessionChannel.sendBlocking(new SessionBindingQueryMessage(address), PacketImpl.SESS_BINDINGQUERY_RESP_V4);
SessionBindingQueryResponseMessage_V4 response = (SessionBindingQueryResponseMessage_V4) packet;
- return new AddressQueryImpl(response.isExists(), response.getQueueNames(), response.isAutoCreateQueues(), response.isAutoCreateAddresses(), response.isDefaultPurgeOnNoConsumers(), response.getDefaultMaxConsumers(), response.isDefaultExclusive(), response.isDefaultLastValue());
+ return new AddressQueryImpl(response.isExists(), response.getQueueNames(), response.isAutoCreateQueues(), response.isAutoCreateAddresses(), response.isDefaultPurgeOnNoConsumers(), response.getDefaultMaxConsumers(), response.isDefaultExclusive(), response.isDefaultLastValue(), response.getDefaultLastValueKey(), response.isDefaultNonDestructive(), response.getDefaultConsumersBeforeDispatch(), response.getDefaultDelayBeforeDispatch());
} else if (sessionChannel.supports(PacketImpl.SESS_BINDINGQUERY_RESP_V3, getServerVersion())) {
Packet packet = sessionChannel.sendBlocking(new SessionBindingQueryMessage(address), PacketImpl.SESS_BINDINGQUERY_RESP_V3);
SessionBindingQueryResponseMessage_V3 response = (SessionBindingQueryResponseMessage_V3) packet;
- return new AddressQueryImpl(response.isExists(), response.getQueueNames(), response.isAutoCreateQueues(), response.isAutoCreateAddresses(), ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), null, null);
+ return new AddressQueryImpl(response.isExists(), response.getQueueNames(), response.isAutoCreateQueues(), response.isAutoCreateAddresses(), ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), null, null, null, null, null, null);
} else if (sessionChannel.supports(PacketImpl.SESS_BINDINGQUERY_RESP_V2, getServerVersion())) {
Packet packet = sessionChannel.sendBlocking(new SessionBindingQueryMessage(address), PacketImpl.SESS_BINDINGQUERY_RESP_V2);
SessionBindingQueryResponseMessage_V2 response = (SessionBindingQueryResponseMessage_V2) packet;
- return new AddressQueryImpl(response.isExists(), response.getQueueNames(), response.isAutoCreateQueues(), false, ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), null, null);
+ return new AddressQueryImpl(response.isExists(), response.getQueueNames(), response.isAutoCreateQueues(), false, ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), null, null, null, null, null, null);
} else {
Packet packet = sessionChannel.sendBlocking(new SessionBindingQueryMessage(address), PacketImpl.SESS_BINDINGQUERY_RESP);
SessionBindingQueryResponseMessage response = (SessionBindingQueryResponseMessage) packet;
- return new AddressQueryImpl(response.isExists(), response.getQueueNames(), false, false, ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), null, null);
+ return new AddressQueryImpl(response.isExists(), response.getQueueNames(), false, false, ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), null, null, null, null, null, null);
}
}
@@ -704,6 +733,21 @@ public class ActiveMQSessionContext extends SessionContext {
@Override
public void createQueue(SimpleString address,
+ SimpleString queueName,
+ boolean temp,
+ boolean autoCreated,
+ QueueAttributes queueAttributes) throws ActiveMQException {
+ if (sessionChannel.getConnection().isVersionBeforeAddressChange()) {
+ CreateQueueMessage request = new CreateQueueMessage(address, queueName, queueAttributes.getFilterString(), queueAttributes.getDurable(), temp, true);
+ sessionChannel.sendBlocking(request, PacketImpl.NULL_RESPONSE);
+ } else {
+ CreateQueueMessage request = new CreateQueueMessage_V2(address, queueName, temp, autoCreated, true, queueAttributes);
+ sessionChannel.sendBlocking(request, PacketImpl.NULL_RESPONSE);
+ }
+ }
+
+ @Override
+ public void createQueue(SimpleString address,
RoutingType routingType,
SimpleString queueName,
SimpleString filterString,
@@ -714,13 +758,19 @@ public class ActiveMQSessionContext extends SessionContext {
boolean autoCreated,
Boolean exclusive,
Boolean lastValue) throws ActiveMQException {
- if (sessionChannel.getConnection().isVersionBeforeAddressChange()) {
- CreateQueueMessage request = new CreateQueueMessage(address, queueName, filterString, durable, temp, true);
- sessionChannel.sendBlocking(request, PacketImpl.NULL_RESPONSE);
- } else {
- CreateQueueMessage request = new CreateQueueMessage_V2(address, queueName, routingType, filterString, durable, temp, maxConsumers, purgeOnNoConsumers, autoCreated, true, exclusive, lastValue);
- sessionChannel.sendBlocking(request, PacketImpl.NULL_RESPONSE);
- }
+ createQueue(
+ address,
+ queueName,
+ temp,
+ autoCreated,
+ new QueueAttributes()
+ .setRoutingType(routingType)
+ .setFilterString(filterString)
+ .setDurable(durable)
+ .setMaxConsumers(maxConsumers)
+ .setPurgeOnNoConsumers(purgeOnNoConsumers)
+ .setExclusive(exclusive)
+ .setLastValue(lastValue));
}
@Deprecated
@@ -820,7 +870,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.isLastValue());
+ 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.isLastValue(), queueInfo.getLastValueKey(), queueInfo.isNonDestructive(), queueInfo.getConsumersBeforeDispatch(), queueInfo.getDelayBeforeDispatch());
sendPacketWithoutLock(sessionChannel, createQueueRequest);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage_V2.java
----------------------------------------------------------------------
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 baccc36..78d869d 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
@@ -17,6 +17,7 @@
package org.apache.activemq.artemis.core.protocol.core.impl.wireformat;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.QueueAttributes;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.utils.BufferHelper;
@@ -35,6 +36,40 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
private Boolean lastValue;
+ private SimpleString lastValueKey;
+
+ private Boolean nonDestructive;
+
+ private Integer consumersBeforeDispatch;
+
+ private Long delayBeforeDispatch;
+
+ public CreateQueueMessage_V2(final SimpleString address,
+ final SimpleString queueName,
+ final boolean temporary,
+ final boolean autoCreated,
+ final boolean requiresResponse,
+ final QueueAttributes queueAttributes) {
+ this(
+ address,
+ queueName,
+ queueAttributes.getRoutingType(),
+ queueAttributes.getFilterString(),
+ queueAttributes.getDurable(),
+ temporary,
+ queueAttributes.getMaxConsumers(),
+ queueAttributes.getPurgeOnNoConsumers(),
+ autoCreated,
+ requiresResponse,
+ queueAttributes.getExclusive(),
+ queueAttributes.getLastValue(),
+ queueAttributes.getLastValueKey(),
+ queueAttributes.getNonDestructive(),
+ queueAttributes.getConsumersBeforeDispatch(),
+ queueAttributes.getDelayBeforeDispatch()
+ );
+ }
+
public CreateQueueMessage_V2(final SimpleString address,
final SimpleString queueName,
final RoutingType routingType,
@@ -46,7 +81,11 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
final boolean autoCreated,
final boolean requiresResponse,
final Boolean exclusive,
- final Boolean lastValue) {
+ final Boolean lastValue,
+ final SimpleString lastValueKey,
+ final Boolean nonDestructive,
+ final Integer consumersBeforeDispatch,
+ final Long delayBeforeDispatch) {
this();
this.address = address;
@@ -61,6 +100,10 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
this.purgeOnNoConsumers = purgeOnNoConsumers;
this.exclusive = exclusive;
this.lastValue = lastValue;
+ this.lastValueKey = lastValueKey;
+ this.nonDestructive = nonDestructive;
+ this.consumersBeforeDispatch = consumersBeforeDispatch;
+ this.delayBeforeDispatch = delayBeforeDispatch;
}
public CreateQueueMessage_V2() {
@@ -78,6 +121,10 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
buff.append(", purgeOnNoConsumers=" + purgeOnNoConsumers);
buff.append(", exclusive=" + exclusive);
buff.append(", lastValue=" + lastValue);
+ buff.append(", lastValueKey=" + lastValue);
+ buff.append(", nonDestructive=" + nonDestructive);
+ buff.append(", consumersBeforeDispatch=" + consumersBeforeDispatch);
+ buff.append(", delayBeforeDispatch=" + delayBeforeDispatch);
buff.append("]");
return buff.toString();
}
@@ -130,6 +177,38 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
this.lastValue = lastValue;
}
+ public SimpleString getLastValueKey() {
+ return lastValueKey;
+ }
+
+ public void setLastValueKey(SimpleString lastValueKey) {
+ this.lastValueKey = lastValueKey;
+ }
+
+ public Boolean isNonDestructive() {
+ return nonDestructive;
+ }
+
+ public void setNonDestructive(Boolean nonDestructive) {
+ this.nonDestructive = nonDestructive;
+ }
+
+ public Integer getConsumersBeforeDispatch() {
+ return consumersBeforeDispatch;
+ }
+
+ public void setConsumersBeforeDispatch(Integer consumersBeforeDispatch) {
+ this.consumersBeforeDispatch = consumersBeforeDispatch;
+ }
+
+ public Long getDelayBeforeDispatch() {
+ return delayBeforeDispatch;
+ }
+
+ public void setDelayBeforeDispatch(Long delayBeforeDispatch) {
+ this.delayBeforeDispatch = delayBeforeDispatch;
+ }
+
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
super.encodeRest(buffer);
@@ -139,6 +218,10 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
buffer.writeBoolean(purgeOnNoConsumers);
BufferHelper.writeNullableBoolean(buffer, exclusive);
BufferHelper.writeNullableBoolean(buffer, lastValue);
+ buffer.writeNullableSimpleString(lastValueKey);
+ BufferHelper.writeNullableBoolean(buffer, nonDestructive);
+ BufferHelper.writeNullableInteger(buffer, consumersBeforeDispatch);
+ BufferHelper.writeNullableLong(buffer, delayBeforeDispatch);
}
@Override
@@ -152,6 +235,12 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
exclusive = BufferHelper.readNullableBoolean(buffer);
lastValue = BufferHelper.readNullableBoolean(buffer);
}
+ if (buffer.readableBytes() > 0) {
+ lastValueKey = buffer.readNullableSimpleString();
+ nonDestructive = BufferHelper.readNullableBoolean(buffer);
+ consumersBeforeDispatch = BufferHelper.readNullableInteger(buffer);
+ delayBeforeDispatch = BufferHelper.readNullableLong(buffer);
+ }
}
@Override
@@ -164,6 +253,10 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
result = prime * result + (purgeOnNoConsumers ? 1231 : 1237);
result = prime * result + (exclusive == null ? 0 : exclusive ? 1231 : 1237);
result = prime * result + (lastValue == null ? 0 : lastValue ? 1231 : 1237);
+ result = prime * result + (lastValueKey == null ? 0 : lastValueKey.hashCode());
+ 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());
return result;
}
@@ -192,6 +285,26 @@ public class CreateQueueMessage_V2 extends CreateQueueMessage {
return false;
} else if (!lastValue.equals(other.lastValue))
return false;
+ if (lastValueKey == null) {
+ if (other.lastValueKey != null)
+ return false;
+ } else if (!lastValueKey.equals(other.lastValueKey))
+ return false;
+ if (nonDestructive == null) {
+ if (other.nonDestructive != null)
+ return false;
+ } else if (!nonDestructive.equals(other.nonDestructive))
+ return false;
+ if (consumersBeforeDispatch == null) {
+ if (other.consumersBeforeDispatch != null)
+ return false;
+ } else if (!consumersBeforeDispatch.equals(other.consumersBeforeDispatch))
+ return false;
+ if (delayBeforeDispatch == null) {
+ if (other.delayBeforeDispatch != null)
+ return false;
+ } else if (!delayBeforeDispatch.equals(other.delayBeforeDispatch))
+ return false;
if (routingType == null) {
if (other.routingType != null)
return false;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage_V2.java
----------------------------------------------------------------------
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 d220915..90ef13b 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
@@ -28,6 +28,10 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
Boolean purgeOnNoConsumers;
private Boolean exclusive;
private Boolean lastValue;
+ private SimpleString lastValueKey;
+ private Boolean nonDestructive;
+ private Integer consumersBeforeDispatch;
+ private Long delayBeforeDispatch;
public CreateSharedQueueMessage_V2(final SimpleString address,
final SimpleString queueName,
@@ -38,6 +42,10 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
final Boolean purgeOnNoConsumers,
final Boolean exclusive,
final Boolean lastValue,
+ final SimpleString lastValueKey,
+ final Boolean nonDestructive,
+ final Integer consumersBeforeDispatch,
+ final Long delayBeforeDispatch,
final boolean requiresResponse) {
this();
@@ -50,8 +58,11 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
this.purgeOnNoConsumers = purgeOnNoConsumers;
this.exclusive = exclusive;
this.lastValue = lastValue;
+ this.lastValueKey = lastValueKey;
+ this.nonDestructive = nonDestructive;
+ this.consumersBeforeDispatch = consumersBeforeDispatch;
+ this.delayBeforeDispatch = delayBeforeDispatch;
this.requiresResponse = requiresResponse;
-
}
public CreateSharedQueueMessage_V2() {
@@ -98,6 +109,38 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
this.lastValue = lastValue;
}
+ public SimpleString getLastValueKey() {
+ return lastValueKey;
+ }
+
+ public void setLastValueKey(SimpleString lastValueKey) {
+ this.lastValueKey = lastValueKey;
+ }
+
+ public Boolean isNonDestructive() {
+ return nonDestructive;
+ }
+
+ public void setNonDestructive(Boolean nonDestructive) {
+ this.nonDestructive = nonDestructive;
+ }
+
+ public Integer getConsumersBeforeDispatch() {
+ return consumersBeforeDispatch;
+ }
+
+ public void setConsumersBeforeDispatch(Integer consumersBeforeDispatch) {
+ this.consumersBeforeDispatch = consumersBeforeDispatch;
+ }
+
+ public Long getDelayBeforeDispatch() {
+ return delayBeforeDispatch;
+ }
+
+ public void setDelayBeforeDispatch(Long delayBeforeDispatch) {
+ this.delayBeforeDispatch = delayBeforeDispatch;
+ }
+
@Override
public String toString() {
StringBuffer buff = new StringBuffer(getParentString());
@@ -110,6 +153,10 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
buff.append(", purgeOnNoConsumers=" + purgeOnNoConsumers);
buff.append(", exclusive=" + exclusive);
buff.append(", lastValue=" + lastValue);
+ buff.append(", lastValueKey=" + lastValueKey);
+ buff.append(", nonDestructive=" + nonDestructive);
+ buff.append(", consumersBeforeDispatch=" + consumersBeforeDispatch);
+ buff.append(", delayBeforeDispatch=" + delayBeforeDispatch);
buff.append(", requiresResponse=" + requiresResponse);
buff.append("]");
return buff.toString();
@@ -127,6 +174,10 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
BufferHelper.writeNullableBoolean(buffer, purgeOnNoConsumers);
BufferHelper.writeNullableBoolean(buffer, exclusive);
BufferHelper.writeNullableBoolean(buffer, lastValue);
+ buffer.writeNullableSimpleString(lastValueKey);
+ BufferHelper.writeNullableBoolean(buffer, nonDestructive);
+ BufferHelper.writeNullableInteger(buffer, consumersBeforeDispatch);
+ BufferHelper.writeNullableLong(buffer, delayBeforeDispatch);
}
@Override
@@ -143,6 +194,12 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
exclusive = BufferHelper.readNullableBoolean(buffer);
lastValue = BufferHelper.readNullableBoolean(buffer);
}
+ if (buffer.readableBytes() > 0) {
+ lastValueKey = buffer.readNullableSimpleString();
+ nonDestructive = BufferHelper.readNullableBoolean(buffer);
+ consumersBeforeDispatch = BufferHelper.readNullableInteger(buffer);
+ delayBeforeDispatch = BufferHelper.readNullableLong(buffer);
+ }
}
@Override
@@ -159,6 +216,11 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
result = prime * result + (purgeOnNoConsumers == null ? 0 : purgeOnNoConsumers ? 1231 : 1237);
result = prime * result + (exclusive == null ? 0 : exclusive ? 1231 : 1237);
result = prime * result + (lastValue == null ? 0 : lastValue ? 1231 : 1237);
+ result = prime * result + (lastValueKey == null ? 0 : lastValueKey.hashCode());
+ 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());
+
return result;
}
@@ -212,6 +274,26 @@ public class CreateSharedQueueMessage_V2 extends CreateSharedQueueMessage {
return false;
} else if (!lastValue.equals(other.lastValue))
return false;
+ if (lastValueKey == null) {
+ if (other.lastValueKey != null)
+ return false;
+ } else if (!lastValueKey.equals(other.lastValueKey))
+ return false;
+ if (nonDestructive == null) {
+ if (other.nonDestructive != null)
+ return false;
+ } else if (!nonDestructive.equals(other.nonDestructive))
+ return false;
+ if (consumersBeforeDispatch == null) {
+ if (other.consumersBeforeDispatch != null)
+ return false;
+ } else if (!consumersBeforeDispatch.equals(other.consumersBeforeDispatch))
+ return false;
+ if (delayBeforeDispatch == null) {
+ if (other.delayBeforeDispatch != null)
+ return false;
+ } else if (!delayBeforeDispatch.equals(other.delayBeforeDispatch))
+ return false;
return true;
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionBindingQueryResponseMessage_V4.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionBindingQueryResponseMessage_V4.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionBindingQueryResponseMessage_V4.java
index ca0e9f7..562a740 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionBindingQueryResponseMessage_V4.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionBindingQueryResponseMessage_V4.java
@@ -32,6 +32,14 @@ public class SessionBindingQueryResponseMessage_V4 extends SessionBindingQueryRe
private Boolean defaultLastValue;
+ private SimpleString defaultLastValueKey;
+
+ private Boolean defaultNonDestructive;
+
+ private Integer defaultConsumersBeforeDispatch;
+
+ private Long defaultDelayBeforeDispatch;
+
public SessionBindingQueryResponseMessage_V4(final boolean exists,
final List<SimpleString> queueNames,
final boolean autoCreateQueues,
@@ -39,7 +47,11 @@ public class SessionBindingQueryResponseMessage_V4 extends SessionBindingQueryRe
final boolean defaultPurgeOnNoConsumers,
final int defaultMaxConsumers,
final Boolean defaultExclusive,
- final Boolean defaultLastValue) {
+ final Boolean defaultLastValue,
+ final SimpleString defaultLastValueKey,
+ final Boolean defaultNonDestructive,
+ final Integer defaultConsumersBeforeDispatch,
+ final Long defaultDelayBeforeDispatch) {
super(SESS_BINDINGQUERY_RESP_V4);
this.exists = exists;
@@ -57,6 +69,14 @@ public class SessionBindingQueryResponseMessage_V4 extends SessionBindingQueryRe
this.defaultExclusive = defaultExclusive;
this.defaultLastValue = defaultLastValue;
+
+ this.defaultLastValueKey = defaultLastValueKey;
+
+ this.defaultNonDestructive = defaultNonDestructive;
+
+ this.defaultConsumersBeforeDispatch = defaultConsumersBeforeDispatch;
+
+ this.defaultDelayBeforeDispatch = defaultDelayBeforeDispatch;
}
public SessionBindingQueryResponseMessage_V4() {
@@ -79,6 +99,22 @@ public class SessionBindingQueryResponseMessage_V4 extends SessionBindingQueryRe
return defaultLastValue;
}
+ public SimpleString getDefaultLastValueKey() {
+ return defaultLastValueKey;
+ }
+
+ public Boolean isDefaultNonDestructive() {
+ return defaultNonDestructive;
+ }
+
+ public Integer getDefaultConsumersBeforeDispatch() {
+ return defaultConsumersBeforeDispatch;
+ }
+
+ public Long getDefaultDelayBeforeDispatch() {
+ return defaultDelayBeforeDispatch;
+ }
+
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
super.encodeRest(buffer);
@@ -86,6 +122,10 @@ public class SessionBindingQueryResponseMessage_V4 extends SessionBindingQueryRe
buffer.writeInt(defaultMaxConsumers);
BufferHelper.writeNullableBoolean(buffer, defaultExclusive);
BufferHelper.writeNullableBoolean(buffer, defaultLastValue);
+ buffer.writeNullableSimpleString(defaultLastValueKey);
+ BufferHelper.writeNullableBoolean(buffer, defaultNonDestructive);
+ BufferHelper.writeNullableInteger(buffer, defaultConsumersBeforeDispatch);
+ BufferHelper.writeNullableLong(buffer, defaultDelayBeforeDispatch);
}
@Override
@@ -97,6 +137,12 @@ public class SessionBindingQueryResponseMessage_V4 extends SessionBindingQueryRe
defaultExclusive = BufferHelper.readNullableBoolean(buffer);
defaultLastValue = BufferHelper.readNullableBoolean(buffer);
}
+ if (buffer.readableBytes() > 0) {
+ defaultLastValueKey = buffer.readNullableSimpleString();
+ defaultNonDestructive = BufferHelper.readNullableBoolean(buffer);
+ defaultConsumersBeforeDispatch = BufferHelper.readNullableInteger(buffer);
+ defaultDelayBeforeDispatch = BufferHelper.readNullableLong(buffer);
+ }
}
@Override
@@ -107,6 +153,10 @@ public class SessionBindingQueryResponseMessage_V4 extends SessionBindingQueryRe
result = prime * result + defaultMaxConsumers;
result = prime * result + (defaultExclusive == null ? 0 : defaultExclusive ? 1231 : 1237);
result = prime * result + (defaultLastValue == null ? 0 : defaultLastValue ? 1231 : 1237);
+ result = prime * result + (defaultLastValueKey == null ? 0 : defaultLastValueKey.hashCode());
+ result = prime * result + (defaultNonDestructive == null ? 0 : defaultNonDestructive ? 1231 : 1237);
+ result = prime * result + (defaultConsumersBeforeDispatch == null ? 0 : defaultConsumersBeforeDispatch.hashCode());
+ result = prime * result + (defaultDelayBeforeDispatch == null ? 0 : defaultDelayBeforeDispatch.hashCode());
return result;
}
@@ -124,6 +174,10 @@ public class SessionBindingQueryResponseMessage_V4 extends SessionBindingQueryRe
buff.append(", defaultMaxConsumers=" + defaultMaxConsumers);
buff.append(", defaultExclusive=" + defaultExclusive);
buff.append(", defaultLastValue=" + defaultLastValue);
+ buff.append(", defaultLastValueKey=" + defaultLastValueKey);
+ buff.append(", defaultNonDestructive=" + defaultNonDestructive);
+ buff.append(", defaultConsumersBeforeDispatch=" + defaultConsumersBeforeDispatch);
+ buff.append(", defaultDelayBeforeDispatch=" + defaultDelayBeforeDispatch);
return buff.toString();
}
@@ -150,6 +204,26 @@ public class SessionBindingQueryResponseMessage_V4 extends SessionBindingQueryRe
return false;
} else if (!defaultLastValue.equals(other.defaultLastValue))
return false;
+ if (defaultLastValueKey == null) {
+ if (other.defaultLastValueKey != null)
+ return false;
+ } else if (!defaultLastValueKey.equals(other.defaultLastValueKey))
+ return false;
+ if (defaultNonDestructive == null) {
+ if (other.defaultNonDestructive != null)
+ return false;
+ } else if (!defaultNonDestructive.equals(other.defaultNonDestructive))
+ return false;
+ if (defaultConsumersBeforeDispatch == null) {
+ if (other.defaultConsumersBeforeDispatch != null)
+ return false;
+ } else if (!defaultConsumersBeforeDispatch.equals(other.defaultConsumersBeforeDispatch))
+ return false;
+ if (defaultDelayBeforeDispatch == null) {
+ if (other.defaultDelayBeforeDispatch != null)
+ return false;
+ } else if (!defaultDelayBeforeDispatch.equals(other.defaultDelayBeforeDispatch))
+ return false;
return true;
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/547b2aa5/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionQueueQueryResponseMessage_V3.java
----------------------------------------------------------------------
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 0c4c40f..7e24186 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
@@ -38,14 +38,22 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
protected Boolean lastValue;
+ protected SimpleString lastValueKey;
+
+ protected Boolean nonDestructive;
+
+ private Integer consumersBeforeDispatch;
+
+ private Long delayBeforeDispatch;
+
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.isLastValue(), result.getDefaultConsumerWindowSize());
+ 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.isLastValue(), result.getLastValueKey(), result.isNonDestructive(), result.getConsumersBeforeDispatch(), result.getDelayBeforeDispatch(), result.getDefaultConsumerWindowSize());
}
public SessionQueueQueryResponseMessage_V3() {
- this(null, null, false, false, null, 0, 0, false, false, false, false, RoutingType.MULTICAST, -1, 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);
}
private SessionQueueQueryResponseMessage_V3(final SimpleString name,
@@ -63,6 +71,10 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
final int maxConsumers,
final Boolean exclusive,
final Boolean lastValue,
+ final SimpleString lastValueKey,
+ final Boolean nonDestructive,
+ final Integer consumersBeforeDispatch,
+ final Long delayBeforeDispatch,
final Integer defaultConsumerWindowSize) {
super(SESS_QUEUEQUERY_RESP_V3);
@@ -96,6 +108,14 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
this.lastValue = lastValue;
+ this.lastValueKey = lastValueKey;
+
+ this.nonDestructive = nonDestructive;
+
+ this.consumersBeforeDispatch = consumersBeforeDispatch;
+
+ this.delayBeforeDispatch = delayBeforeDispatch;
+
this.defaultConsumerWindowSize = defaultConsumerWindowSize;
}
@@ -147,6 +167,38 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
this.lastValue = lastValue;
}
+ public SimpleString getLastValueKey() {
+ return lastValueKey;
+ }
+
+ public void setLastValueKey(SimpleString lastValueKey) {
+ this.lastValueKey = lastValueKey;
+ }
+
+ public Boolean isNonDestructive() {
+ return nonDestructive;
+ }
+
+ public void setNonDestructive(Boolean nonDestructive) {
+ this.nonDestructive = nonDestructive;
+ }
+
+ public Integer getConsumersBeforeDispatch() {
+ return consumersBeforeDispatch;
+ }
+
+ public void setConsumersBeforeDispatch(Integer consumersBeforeDispatch) {
+ this.consumersBeforeDispatch = consumersBeforeDispatch;
+ }
+
+ public Long getDelayBeforeDispatch() {
+ return delayBeforeDispatch;
+ }
+
+ public void setDelayBeforeDispatch(Long delayBeforeDispatch) {
+ this.delayBeforeDispatch = delayBeforeDispatch;
+ }
+
public Integer getDefaultConsumerWindowSize() {
return defaultConsumerWindowSize;
}
@@ -165,6 +217,10 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
BufferHelper.writeNullableBoolean(buffer, exclusive);
BufferHelper.writeNullableBoolean(buffer, lastValue);
BufferHelper.writeNullableInteger(buffer, defaultConsumerWindowSize);
+ buffer.writeNullableSimpleString(lastValueKey);
+ BufferHelper.writeNullableBoolean(buffer, nonDestructive);
+ BufferHelper.writeNullableInteger(buffer, consumersBeforeDispatch);
+ BufferHelper.writeNullableLong(buffer, delayBeforeDispatch);
}
@Override
@@ -181,6 +237,12 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
if (buffer.readableBytes() > 0) {
defaultConsumerWindowSize = BufferHelper.readNullableInteger(buffer);
}
+ if (buffer.readableBytes() > 0) {
+ lastValueKey = buffer.readNullableSimpleString();
+ nonDestructive = BufferHelper.readNullableBoolean(buffer);
+ consumersBeforeDispatch = BufferHelper.readNullableInteger(buffer);
+ delayBeforeDispatch = BufferHelper.readNullableLong(buffer);
+ }
}
@Override
@@ -193,6 +255,10 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
result = prime * result + maxConsumers;
result = prime * result + (exclusive == null ? 0 : exclusive ? 1231 : 1237);
result = prime * result + (lastValue == null ? 0 : lastValue ? 1231 : 1237);
+ result = prime * result + (lastValueKey == null ? 0 : lastValueKey.hashCode());
+ 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 + ((defaultConsumerWindowSize == null) ? 0 : defaultConsumerWindowSize.hashCode());
return result;
}
@@ -213,13 +279,17 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
buff.append(", maxConsumers=" + maxConsumers);
buff.append(", exclusive=" + exclusive);
buff.append(", lastValue=" + lastValue);
+ buff.append(", lastValueKey=" + lastValueKey);
+ buff.append(", nonDestructive=" + nonDestructive);
+ buff.append(", consumersBeforeDispatch=" + consumersBeforeDispatch);
+ buff.append(", delayBeforeDispatch=" + delayBeforeDispatch);
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(), isLastValue(), getDefaultConsumerWindowSize());
+ return new QueueQueryImpl(isDurable(), isTemporary(), getConsumerCount(), getMessageCount(), getFilterString(), getAddress(), getName(), isExists(), isAutoCreateQueues(), getMaxConsumers(), isAutoCreated(), isPurgeOnNoConsumers(), getRoutingType(), isExclusive(), isLastValue(), getLastValueKey(), isNonDestructive(), getConsumersBeforeDispatch(), getDelayBeforeDispatch(), getDefaultConsumerWindowSize());
}
@Override
@@ -245,6 +315,26 @@ public class SessionQueueQueryResponseMessage_V3 extends SessionQueueQueryRespon
return false;
} else if (!lastValue.equals(other.lastValue))
return false;
+ if (lastValueKey == null) {
+ if (other.lastValueKey != null)
+ return false;
+ } else if (!lastValueKey.equals(other.lastValueKey))
+ return false;
+ if (nonDestructive == null) {
+ if (other.nonDestructive != null)
+ return false;
+ } else if (!nonDestructive.equals(other.nonDestructive))
+ return false;
+ if (consumersBeforeDispatch == null) {
+ if (other.consumersBeforeDispatch != null)
+ return false;
+ } else if (!consumersBeforeDispatch.equals(other.consumersBeforeDispatch))
+ return false;
+ if (delayBeforeDispatch == null) {
+ if (other.delayBeforeDispatch != null)
+ return false;
+ } else if (!delayBeforeDispatch.equals(other.delayBeforeDispatch))
+ return false;
if (defaultConsumerWindowSize == null) {
if (other.defaultConsumerWindowSize != null)
return false;