http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQSession.java ---------------------------------------------------------------------- diff --git a/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQSession.java b/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQSession.java index cf2ec59..a0ccfcc 100644 --- a/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQSession.java +++ b/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQSession.java @@ -52,6 +52,7 @@ import java.util.concurrent.ConcurrentHashMap; import org.apache.activemq.artemis.api.core.ActiveMQException; import org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException; +import org.apache.activemq.artemis.api.core.QueueAttributes; 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.ClientProducer; @@ -318,7 +319,7 @@ public class ActiveMQSession implements QueueSession, TopicSession { if (jbd.isQueue() && response.isAutoCreateQueues()) { // perhaps just relying on the broker to do it is simplest (i.e. purgeOnNoConsumers) session.createAddress(jbd.getSimpleAddress(), RoutingType.ANYCAST, true); - session.createQueue(jbd.getSimpleAddress(), RoutingType.ANYCAST, jbd.getSimpleAddress(), null, true, true, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers()); + createQueue(jbd, RoutingType.ANYCAST, jbd.getSimpleAddress(), null, true, true, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers(), response.isDefaultExclusive(), response.isDefaultLastValueQueue()); } else if (!jbd.isQueue() && response.isAutoCreateAddresses()) { session.createAddress(jbd.getSimpleAddress(), RoutingType.MULTICAST, true); } else { @@ -629,16 +630,16 @@ public class ActiveMQSession implements QueueSession, TopicSession { queueName = ActiveMQDestination.createQueueNameForSubscription(durability == ConsumerDurability.DURABLE, connection.getClientID(), subscriptionName); - if (durability == ConsumerDurability.DURABLE) { - try { - session.createSharedQueue(dest.getSimpleAddress(), RoutingType.MULTICAST, queueName, coreFilterString, true); - } catch (ActiveMQQueueExistsException ignored) { - // We ignore this because querying and then creating the queue wouldn't be idempotent - // we could also add a parameter to ignore existence what would require a bigger work around to avoid - // compatibility. + try { + if (durability == ConsumerDurability.DURABLE) { + createSharedQueue(dest, RoutingType.MULTICAST, queueName, coreFilterString, true, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers(), response.isDefaultExclusive(), response.isDefaultLastValueQueue()); + } else { + createSharedQueue(dest, RoutingType.MULTICAST, queueName, coreFilterString, false, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers(), response.isDefaultExclusive(), response.isDefaultLastValueQueue()); } - } else { - session.createSharedQueue(dest.getSimpleAddress(), queueName, coreFilterString, false); + } catch (ActiveMQQueueExistsException ignored) { + // We ignore this because querying and then creating the queue wouldn't be idempotent + // we could also add a parameter to ignore existence what would require a bigger work around to avoid + // compatibility. } consumer = session.createConsumer(queueName, null, false); @@ -699,7 +700,7 @@ public class ActiveMQSession implements QueueSession, TopicSession { if (!response.isExists() || !response.getQueueNames().contains(dest.getSimpleAddress())) { if (response.isAutoCreateQueues()) { try { - session.createQueue(dest.getSimpleAddress(), RoutingType.ANYCAST, dest.getSimpleAddress(), null, true, true, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers()); + createQueue(dest, RoutingType.ANYCAST, dest.getSimpleAddress(), null, true, true, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers(), response.isDefaultExclusive(), response.isDefaultLastValueQueue()); } catch (ActiveMQQueueExistsException e) { // The queue was created by another client/admin between the query check and send create queue packet } @@ -733,7 +734,7 @@ public class ActiveMQSession implements QueueSession, TopicSession { queueName = new SimpleString(UUID.randomUUID().toString()); - session.createTemporaryQueue(dest.getSimpleAddress(), RoutingType.MULTICAST, queueName, coreFilterString); + createTemporaryQueue(dest, RoutingType.MULTICAST, queueName, coreFilterString, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers(), response.isDefaultExclusive(), response.isDefaultLastValueQueue()); consumer = session.createConsumer(queueName, null, false); @@ -756,7 +757,7 @@ public class ActiveMQSession implements QueueSession, TopicSession { if (!subResponse.isExists()) { // durable subscription queues are not technically considered to be auto-created - session.createQueue(dest.getSimpleAddress(), RoutingType.MULTICAST, queueName, coreFilterString, true, false, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers()); + createQueue(dest, RoutingType.MULTICAST, queueName, coreFilterString, true, false, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers(), response.isDefaultExclusive(), response.isDefaultLastValueQueue()); } else { // Already exists if (subResponse.getConsumerCount() > 0) { @@ -787,7 +788,7 @@ public class ActiveMQSession implements QueueSession, TopicSession { session.deleteQueue(queueName); // Create the new one - session.createQueue(dest.getSimpleAddress(), RoutingType.MULTICAST, queueName, coreFilterString, true, false, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers()); + createQueue(dest, RoutingType.MULTICAST, queueName, coreFilterString, true, false, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers(), response.isDefaultExclusive(), response.isDefaultLastValueQueue()); } } @@ -849,7 +850,7 @@ public class ActiveMQSession implements QueueSession, TopicSession { AddressQuery response = session.addressQuery(new SimpleString(activeMQDestination.getAddress())); if (!response.isExists()) { if (response.isAutoCreateQueues()) { - session.createQueue(activeMQDestination.getSimpleAddress(), RoutingType.ANYCAST, activeMQDestination.getSimpleAddress(), null, true, true, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers()); + createQueue(activeMQDestination, RoutingType.ANYCAST, activeMQDestination.getSimpleAddress(), null, true, true, response.getDefaultMaxConsumers(), response.isDefaultPurgeOnNoConsumers(), response.isDefaultExclusive(), response.isDefaultLastValueQueue()); } else { throw new InvalidDestinationException("Destination " + activeMQDestination.getName() + " does not exist"); } @@ -1154,6 +1155,63 @@ public class ActiveMQSession implements QueueSession, TopicSession { } } + private void createTemporaryQueue(ActiveMQDestination destination, RoutingType routingType, SimpleString queueName, SimpleString filter, int maxConsumers, boolean purgeOnNoConsumers, Boolean exclusive, Boolean lastValue) throws ActiveMQException { + QueueAttributes queueAttributes = destination.getQueueAttributes(); + if (queueAttributes == null) { + session.createTemporaryQueue(destination.getSimpleAddress(), routingType, queueName, filter, maxConsumers, purgeOnNoConsumers, exclusive, lastValue); + } else { + session.createTemporaryQueue( + destination.getSimpleAddress(), + routingType, + queueName, + filter, + queueAttributes.getMaxConsumers() == null ? maxConsumers : queueAttributes.getMaxConsumers(), + queueAttributes.getPurgeOnNoConsumers() == null ? purgeOnNoConsumers : queueAttributes.getPurgeOnNoConsumers(), + queueAttributes.getExclusive() == null ? exclusive : queueAttributes.getExclusive(), + queueAttributes.getLastValue() == null ? lastValue : queueAttributes.getLastValue() + ); + } + } + + private void createSharedQueue(ActiveMQDestination destination, RoutingType routingType, SimpleString queueName, SimpleString filter, boolean durable, int maxConsumers, boolean purgeOnNoConsumers, Boolean exclusive, Boolean lastValue) throws ActiveMQException { + QueueAttributes queueAttributes = destination.getQueueAttributes(); + if (queueAttributes == null) { + session.createSharedQueue(destination.getSimpleAddress(), routingType, queueName, filter, durable, maxConsumers, purgeOnNoConsumers, exclusive, lastValue); + } else { + session.createSharedQueue( + destination.getSimpleAddress(), + routingType, + queueName, + filter, + durable, + queueAttributes.getMaxConsumers() == null ? maxConsumers : queueAttributes.getMaxConsumers(), + queueAttributes.getPurgeOnNoConsumers() == null ? purgeOnNoConsumers : queueAttributes.getPurgeOnNoConsumers(), + queueAttributes.getExclusive() == null ? exclusive : queueAttributes.getExclusive(), + queueAttributes.getLastValue() == null ? lastValue : queueAttributes.getLastValue() + ); + } + } + + private void createQueue(ActiveMQDestination destination, RoutingType routingType, SimpleString queueName, SimpleString filter, boolean durable, boolean autoCreated, int maxConsumers, boolean purgeOnNoConsumers, Boolean exclusive, Boolean lastValue) throws ActiveMQException { + QueueAttributes queueAttributes = destination.getQueueAttributes(); + if (queueAttributes == null) { + session.createQueue(destination.getSimpleAddress(), routingType, queueName, filter, durable, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, lastValue); + } else { + session.createQueue( + destination.getSimpleAddress(), + routingType, + queueName, + filter, + durable, + autoCreated, + queueAttributes.getMaxConsumers() == null ? maxConsumers : queueAttributes.getMaxConsumers(), + queueAttributes.getPurgeOnNoConsumers() == null ? purgeOnNoConsumers : queueAttributes.getPurgeOnNoConsumers(), + queueAttributes.getExclusive() == null ? exclusive : queueAttributes.getExclusive(), + queueAttributes.getLastValue() == null ? lastValue : queueAttributes.getLastValue() + ); + } + } + // Inner classes ------------------------------------------------- }
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-jms-client/src/test/java/org/apache/activemq/artemis/jms/client/ActiveMQParameterTest.java ---------------------------------------------------------------------- diff --git a/artemis-jms-client/src/test/java/org/apache/activemq/artemis/jms/client/ActiveMQParameterTest.java b/artemis-jms-client/src/test/java/org/apache/activemq/artemis/jms/client/ActiveMQParameterTest.java new file mode 100644 index 0000000..cf42fe3 --- /dev/null +++ b/artemis-jms-client/src/test/java/org/apache/activemq/artemis/jms/client/ActiveMQParameterTest.java @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.artemis.jms.client; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; + +/** + * Test Address queue parameters are correctly read. + */ +public class ActiveMQParameterTest { + + @Test + public void testQueueParameters() { + ActiveMQDestination activeMQDestination = new ActiveMQQueue("jms.queue.foo?exclusive=true"); + assertTrue(activeMQDestination.getQueueAttributes().getExclusive()); + + assertEquals("jms.queue.foo", activeMQDestination.getAddress()); + + activeMQDestination = new ActiveMQQueue("jms.queue.foo?exclusive=false"); + assertFalse(activeMQDestination.getQueueAttributes().getExclusive()); + + activeMQDestination = new ActiveMQQueue("jms.queue.foo?last-value=true"); + assertTrue(activeMQDestination.getQueueAttributes().getLastValue()); + + activeMQDestination = new ActiveMQQueue("jms.queue.foo?last-value=false"); + assertFalse(activeMQDestination.getQueueAttributes().getLastValue()); + } +} http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-protocols/artemis-hqclient-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/hornetq/client/HornetQClientSessionContext.java ---------------------------------------------------------------------- diff --git a/artemis-protocols/artemis-hqclient-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/hornetq/client/HornetQClientSessionContext.java b/artemis-protocols/artemis-hqclient-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/hornetq/client/HornetQClientSessionContext.java index 185feaf..ee3520f 100644 --- a/artemis-protocols/artemis-hqclient-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/hornetq/client/HornetQClientSessionContext.java +++ b/artemis-protocols/artemis-hqclient-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/hornetq/client/HornetQClientSessionContext.java @@ -72,7 +72,7 @@ public class HornetQClientSessionContext extends ActiveMQSessionContext { public ClientSession.AddressQuery addressQuery(final SimpleString address) throws ActiveMQException { SessionBindingQueryResponseMessage response = (SessionBindingQueryResponseMessage) getSessionChannel().sendBlocking(new SessionBindingQueryMessage(address), PacketImpl.SESS_BINDINGQUERY_RESP); - return new AddressQueryImpl(response.isExists(), response.getQueueNames(), false, false, ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers()); + return new AddressQueryImpl(response.isExists(), response.getQueueNames(), false, false, ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), ActiveMQDefaultConfiguration.getDefaultExclusive(), ActiveMQDefaultConfiguration.getDefaultLastValue()); } @Override http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreQueueConfiguration.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreQueueConfiguration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreQueueConfiguration.java index 6648b7e..236fac6 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreQueueConfiguration.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/CoreQueueConfiguration.java @@ -35,6 +35,10 @@ public class CoreQueueConfiguration implements Serializable { private String user = null; + private Boolean exclusive; + + private Boolean lastValue; + private Integer maxConsumers = ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(); private Boolean purgeOnNoConsumers = ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(); @@ -64,6 +68,14 @@ public class CoreQueueConfiguration implements Serializable { return user; } + public Boolean isExclusive() { + return exclusive; + } + + public Boolean isLastValue() { + return lastValue; + } + /** * @param address the address to set */ @@ -120,6 +132,16 @@ public class CoreQueueConfiguration implements Serializable { return this; } + public CoreQueueConfiguration setExclusive(Boolean exclusive) { + this.exclusive = exclusive; + return this; + } + + public CoreQueueConfiguration setLastValue(Boolean lastValue) { + this.lastValue = lastValue; + return this; + } + public boolean getPurgeOnNoConsumers() { return purgeOnNoConsumers; } @@ -147,6 +169,8 @@ public class CoreQueueConfiguration implements Serializable { result = prime * result + ((name == null) ? 0 : name.hashCode()); result = prime * result + ((maxConsumers == null) ? 0 : maxConsumers.hashCode()); result = prime * result + ((purgeOnNoConsumers == null) ? 0 : purgeOnNoConsumers.hashCode()); + result = prime * result + ((exclusive == null) ? 0 : exclusive.hashCode()); + result = prime * result + ((lastValue == null) ? 0 : lastValue.hashCode()); return result; } @@ -187,6 +211,18 @@ public class CoreQueueConfiguration implements Serializable { } else if (!purgeOnNoConsumers.equals(other.purgeOnNoConsumers)) { return false; } + if (exclusive == null) { + if (other.exclusive != null) + return false; + } else if (!exclusive.equals(other.exclusive)) { + return false; + } + if (lastValue == null) { + if (other.lastValue != null) + return false; + } else if (!lastValue.equals(other.lastValue)) { + return false; + } return true; } } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java index 8eee4dc..b5352ba 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 @@ -179,6 +179,10 @@ public final class FileConfigurationParser extends XMLConfigurationUtil { private static final String LVQ_NODE_NAME = "last-value-queue"; + private static final String DEFAULT_LVQ_NODE_NAME = "default-last-value-queue"; + + private static final String DEFAULT_EXCLUSIVE_NODE_NAME = "default-exclusive-queue"; + private static final String REDISTRIBUTION_DELAY_NODE_NAME = "redistribution-delay"; private static final String SEND_TO_DLA_ON_NO_ROUTE = "send-to-dla-on-no-route"; @@ -989,8 +993,10 @@ public final class FileConfigurationParser extends XMLConfigurationUtil { Validators.ADDRESS_FULL_MESSAGE_POLICY_TYPE.validate(ADDRESS_FULL_MESSAGE_POLICY_NODE_NAME, value); AddressFullMessagePolicy policy = Enum.valueOf(AddressFullMessagePolicy.class, value); addressSettings.setAddressFullMessagePolicy(policy); - } else if (LVQ_NODE_NAME.equalsIgnoreCase(name)) { - addressSettings.setLastValueQueue(XMLUtil.parseBoolean(child)); + } else if (LVQ_NODE_NAME.equalsIgnoreCase(name) || DEFAULT_LVQ_NODE_NAME.equalsIgnoreCase(name)) { + addressSettings.setDefaultLastValueQueue(XMLUtil.parseBoolean(child)); + } else if (DEFAULT_EXCLUSIVE_NODE_NAME.equalsIgnoreCase(name)) { + addressSettings.setDefaultExclusiveQueue(XMLUtil.parseBoolean(child)); } else if (MAX_DELIVERY_ATTEMPTS.equalsIgnoreCase(name)) { addressSettings.setMaxDeliveryAttempts(XMLUtil.parseInt(child)); } else if (REDISTRIBUTION_DELAY_NODE_NAME.equalsIgnoreCase(name)) { @@ -1090,6 +1096,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil { int maxConsumers = ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(); boolean purgeOnNoConsumers = ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(); String user = null; + Boolean exclusive = null; + Boolean lastValue = null; NamedNodeMap attributes = node.getAttributes(); for (int i = 0; i < attributes.getLength(); i++) { @@ -1099,6 +1107,10 @@ public final class FileConfigurationParser extends XMLConfigurationUtil { Validators.MAX_QUEUE_CONSUMERS.validate(name, maxConsumers); } else if (item.getNodeName().equals("purge-on-no-consumers")) { purgeOnNoConsumers = Boolean.parseBoolean(item.getNodeValue()); + } else if (item.getNodeName().equals("exclusive")) { + exclusive = Boolean.parseBoolean(item.getNodeValue()); + } else if (item.getNodeName().equals("last-value")) { + lastValue = Boolean.parseBoolean(item.getNodeValue()); } } @@ -1117,7 +1129,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil { } } - return new CoreQueueConfiguration().setAddress(address).setName(name).setFilterString(filterString).setDurable(durable).setMaxConsumers(maxConsumers).setPurgeOnNoConsumers(purgeOnNoConsumers).setUser(user); + return new CoreQueueConfiguration().setAddress(address).setName(name).setFilterString(filterString).setDurable(durable).setMaxConsumers(maxConsumers).setPurgeOnNoConsumers(purgeOnNoConsumers).setUser(user) + .setExclusive(exclusive).setLastValue(lastValue); } protected CoreAddressConfiguration parseAddressConfiguration(final Node node) { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java index 4e80bb5..41d7b30 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 @@ -823,12 +823,21 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active String routingType, Integer maxConsumers, Boolean purgeOnNoConsumers) throws Exception { + return updateQueue(name, routingType, maxConsumers, purgeOnNoConsumers, null); + } + + @Override + public String updateQueue(String name, + String routingType, + Integer maxConsumers, + Boolean purgeOnNoConsumers, + Boolean exclusive) throws Exception { checkStarted(); clearIO(); try { - final Queue queue = server.updateQueue(name, routingType != null ? RoutingType.valueOf(routingType) : null, maxConsumers, purgeOnNoConsumers); + final Queue queue = server.updateQueue(name, routingType != null ? RoutingType.valueOf(routingType) : null, maxConsumers, purgeOnNoConsumers, exclusive); if (queue == null) { throw ActiveMQMessageBundle.BUNDLE.noSuchQueue(new SimpleString(name)); } @@ -2077,7 +2086,7 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active .add("redeliveryMultiplier", addressSettings.getRedeliveryMultiplier()) .add("maxRedeliveryDelay", addressSettings.getMaxRedeliveryDelay()) .add("redistributionDelay", addressSettings.getRedistributionDelay()) - .add("lastValueQueue", addressSettings.isLastValueQueue()) + .add("lastValueQueue", addressSettings.isDefaultLastValueQueue()) .add("sendToDLAOnNoRoute", addressSettings.isSendToDLAOnNoRoute()) .add("addressFullMessagePolicy", policy) .add("slowConsumerThreshold", addressSettings.getSlowConsumerThreshold()) @@ -2163,7 +2172,7 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active addressSettings.setDeadLetterAddress(DLA == null ? null : new SimpleString(DLA)); addressSettings.setExpiryAddress(expiryAddress == null ? null : new SimpleString(expiryAddress)); addressSettings.setExpiryDelay(expiryDelay); - addressSettings.setLastValueQueue(lastValueQueue); + addressSettings.setDefaultLastValueQueue(lastValueQueue); addressSettings.setMaxDeliveryAttempts(deliveryAttempts); addressSettings.setPageCacheMaxSize(pageMaxCacheSize); addressSettings.setMaxSizeBytes(maxSizeBytes); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java index 0a1c385..cefcbf9 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java @@ -382,6 +382,30 @@ public class QueueControlImpl extends AbstractControl implements QueueControl { } @Override + public boolean isExclusive() { + checkStarted(); + + clearIO(); + try { + return queue.isExclusive(); + } finally { + blockOnIO(); + } + } + + @Override + public boolean isLastValue() { + checkStarted(); + + clearIO(); + try { + return queue.isLastValue(); + } finally { + blockOnIO(); + } + } + + @Override public Map<String, Object>[] listScheduledMessages() throws Exception { checkStarted(); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/QueueView.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/QueueView.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/QueueView.java index eb0f1a4..181b6ce 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/QueueView.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/QueueView.java @@ -60,7 +60,9 @@ public class QueueView extends ActiveMQAbstractView<QueueControl> { .add("messagesAcked", toString(queue.getMessagesAcknowledged())) .add("deliveringCount", toString(queue.getDeliveringCount())) .add("messagesKilled", toString(queue.getMessagesKilled())) - .add("deliverDeliver", toString(q.isDirectDeliver())); + .add("deliverDeliver", toString(q.isDirectDeliver())) + .add("exclusive", toString(queue.isExclusive())) + .add("lastValue", toString(queue.isLastValue())); return obj; } @@ -108,6 +110,10 @@ public class QueueView extends ActiveMQAbstractView<QueueControl> { return queue.getMessagesKilled(); case "deliverDeliver": return q.isDirectDeliver(); + case "exclusive": + return q.isExclusive(); + case "lastValue": + return q.isLastValue(); default: throw new IllegalArgumentException("Unsupported field, " + fieldName); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/predicate/QueueFilterPredicate.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/predicate/QueueFilterPredicate.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/predicate/QueueFilterPredicate.java index d8209a7..10a0e91 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/predicate/QueueFilterPredicate.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/predicate/QueueFilterPredicate.java @@ -27,7 +27,7 @@ public class QueueFilterPredicate extends ActiveMQFilterPredicate<QueueControl> enum Field { ID, NAME, CONSUMER_ID, QUEUE, ADDRESS, MAX_CONSUMERS, FILTER, MESSAGE_COUNT, CONSUMER_COUNT, DELIVERING_COUNT, MESSAGES_ADDED, MESSAGES_ACKED, RATE, ROUTING_TYPE, USER, AUTO_CREATED, DURABLE, PAUSED, TEMPORARY, - PURGE_ON_NO_CONSUMERS, MESSAGES_KILLED, DIRECT_DELIVER + PURGE_ON_NO_CONSUMERS, MESSAGES_KILLED, DIRECT_DELIVER, LAST_VALUE, EXCLUSIVE } private Field f; @@ -89,6 +89,10 @@ public class QueueFilterPredicate extends ActiveMQFilterPredicate<QueueControl> return matches(queue.isPurgeOnNoConsumers()); case MESSAGES_KILLED: return matches(queue.getMessagesKilled()); + case EXCLUSIVE: + return matches(queue.isExclusive()); + case LAST_VALUE: + return matches(queue.isLastValue()); default: return true; } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/QueueBindingInfo.java index 7164cc1..00b7974 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 @@ -54,6 +54,10 @@ public interface QueueBindingInfo { void setPurgeOnNoConsumers(boolean purgeOnNoConsumers); + boolean isExclusive(); + + void setExclusive(boolean exclusive); + byte getRoutingType(); void setRoutingType(byte routingType); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java index 5441368..34d249e 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 @@ -1272,7 +1272,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.getRoutingType().getType()); + PersistentQueueBindingEncoding bindingEncoding = new PersistentQueueBindingEncoding(queue.getName(), binding.getAddress(), filterString, queue.getUser(), queue.isAutoCreated(), queue.getMaxConsumers(), queue.isPurgeOnNoConsumers(), queue.isExclusive(), queue.getRoutingType().getType()); readLock(); try { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PersistentQueueBindingEncoding.java index af2c675..d67b300 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 @@ -46,6 +46,8 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin public boolean purgeOnNoConsumers; + public boolean exclusive; + public byte routingType; public PersistentQueueBindingEncoding() { @@ -68,6 +70,8 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin maxConsumers + ", purgeOnNoConsumers=" + purgeOnNoConsumers + + ", exclusive=" + + exclusive + ", routingType=" + routingType + "]"; @@ -80,6 +84,7 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin final boolean autoCreated, final int maxConsumers, final boolean purgeOnNoConsumers, + final boolean exclusive, final byte routingType) { this.name = name; this.address = address; @@ -88,6 +93,7 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin this.autoCreated = autoCreated; this.maxConsumers = maxConsumers; this.purgeOnNoConsumers = purgeOnNoConsumers; + this.exclusive = exclusive; this.routingType = routingType; } @@ -164,6 +170,16 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin } @Override + public boolean isExclusive() { + return exclusive; + } + + @Override + public void setExclusive(boolean exclusive) { + this.exclusive = exclusive; + } + + @Override public byte getRoutingType() { return routingType; } @@ -203,6 +219,12 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin purgeOnNoConsumers = ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(); routingType = ActiveMQDefaultConfiguration.getDefaultRoutingType().getType(); } + + if (buffer.readableBytes() > 0) { + exclusive = buffer.readBoolean(); + } else { + exclusive = ActiveMQDefaultConfiguration.getDefaultExclusive(); + } } @Override @@ -215,6 +237,7 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin buffer.writeInt(maxConsumers); buffer.writeBoolean(purgeOnNoConsumers); buffer.writeByte(routingType); + buffer.writeBoolean(exclusive); } @Override @@ -224,7 +247,8 @@ public class PersistentQueueBindingEncoding implements EncodingSupport, QueueBin SimpleString.sizeofNullableString(createMetadata()) + DataConstants.SIZE_INT + DataConstants.SIZE_BOOLEAN + - DataConstants.SIZE_BYTE; + DataConstants.SIZE_BYTE + + DataConstants.SIZE_BOOLEAN; } private SimpleString createMetadata() { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java index 2a8764e..b78883f 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java @@ -64,7 +64,8 @@ public interface PostOffice extends ActiveMQComponent { QueueBinding updateQueue(SimpleString name, RoutingType routingType, Integer maxConsumers, - Boolean purgeOnNoConsumers) throws Exception; + Boolean purgeOnNoConsumers, + Boolean exclusive) throws Exception; List<Queue> listQueuesForAddress(SimpleString address) throws Exception; http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java index be4a0d5..9de5dcc 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 @@ -464,7 +464,8 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding public QueueBinding updateQueue(SimpleString name, RoutingType routingType, Integer maxConsumers, - Boolean purgeOnNoConsumers) throws Exception { + Boolean purgeOnNoConsumers, + Boolean exclusive) throws Exception { synchronized (addressLock) { final QueueBinding queueBinding = (QueueBinding) addressManager.getBinding(name); if (queueBinding == null) { @@ -504,6 +505,10 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding changed = true; queue.setPurgeOnNoConsumers(purgeOnNoConsumers); } + if (exclusive != null && queue.isExclusive() != exclusive.booleanValue()) { + changed = true; + queue.setExclusive(exclusive); + } if (changed) { final long txID = storageManager.generateID(); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java ---------------------------------------------------------------------- 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 d47fbff..7b129ad 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 @@ -353,7 +353,7 @@ 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.isAutoCreated()); + request.isExclusive(), request.isLastValue(), request.isAutoCreated()); if (requiresResponse) { response = new NullResponseMessage(); } @@ -371,7 +371,7 @@ public class ServerSessionPacketHandler implements ChannelHandler { case CREATE_SHARED_QUEUE_V2: { CreateSharedQueueMessage_V2 request = (CreateSharedQueueMessage_V2) packet; requiresResponse = request.isRequiresResponse(); - session.createSharedQueue(request.getAddress(), request.getQueueName(), request.getRoutingType(), request.isDurable(), request.getFilterString()); + session.createSharedQueue(request.getAddress(), request.getQueueName(), request.getRoutingType(), request.getFilterString(), request.isDurable(), request.getMaxConsumers(), request.isPurgeOnNoConsumers(), request.isExclusive(), request.isLastValue()); if (requiresResponse) { response = new NullResponseMessage(); } @@ -417,13 +417,13 @@ public class ServerSessionPacketHandler implements ChannelHandler { if (!queueNames.isEmpty()) { final List<SimpleString> convertedQueueNames = request.convertQueueNames(clientVersion, queueNames); if (convertedQueueNames != queueNames) { - result = new BindingQueryResult(result.isExists(), result.getAddressInfo(), convertedQueueNames, result.isAutoCreateQueues(), result.isAutoCreateAddresses(), result.isDefaultPurgeOnNoConsumers(), result.getDefaultMaxConsumers()); + result = new BindingQueryResult(result.isExists(), result.getAddressInfo(), convertedQueueNames, result.isAutoCreateQueues(), result.isAutoCreateAddresses(), result.isDefaultPurgeOnNoConsumers(), result.getDefaultMaxConsumers(), result.isDefaultExclusive(), result.isDefaultLastValue()); } } } if (channel.supports(PacketImpl.SESS_BINDINGQUERY_RESP_V4)) { - response = new SessionBindingQueryResponseMessage_V4(result.isExists(), result.getQueueNames(), result.isAutoCreateQueues(), result.isAutoCreateAddresses(), result.isDefaultPurgeOnNoConsumers(), result.getDefaultMaxConsumers()); + response = new SessionBindingQueryResponseMessage_V4(result.isExists(), result.getQueueNames(), result.isAutoCreateQueues(), result.isAutoCreateAddresses(), result.isDefaultPurgeOnNoConsumers(), result.getDefaultMaxConsumers(), result.isDefaultExclusive(), result.isDefaultLastValue()); } else if (channel.supports(PacketImpl.SESS_BINDINGQUERY_RESP_V3)) { response = new SessionBindingQueryResponseMessage_V3(result.isExists(), result.getQueueNames(), result.isAutoCreateQueues(), result.isAutoCreateAddresses()); } else if (channel.supports(PacketImpl.SESS_BINDINGQUERY_RESP_V2)) { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java index 6af94ff..b8598b8 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 @@ -333,6 +333,9 @@ public interface ActiveMQServer extends ServiceComponent { void createSharedQueue(SimpleString address, RoutingType routingType, SimpleString name, SimpleString filterString, SimpleString user, boolean durable) throws Exception; + void createSharedQueue(SimpleString address, RoutingType routingType, SimpleString name, SimpleString filterString, + SimpleString user, boolean durable, int maxConsumers, boolean purgeOnNoConsumers, boolean exclusive, boolean lastValue) throws Exception; + Queue createQueue(SimpleString address, RoutingType routingType, SimpleString queueName, SimpleString filter, boolean durable, boolean temporary) throws Exception; @@ -351,10 +354,18 @@ public interface ActiveMQServer extends ServiceComponent { SimpleString user, boolean durable, boolean temporary, boolean autoCreated, Integer maxConsumers, Boolean purgeOnNoConsumers, boolean autoCreateAddress) throws Exception; + 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; + Queue createQueue(SimpleString address, RoutingType routingType, SimpleString queueName, SimpleString filter, SimpleString user, boolean durable, boolean temporary, boolean ignoreIfExists, boolean transientQueue, boolean autoCreated, int maxConsumers, boolean purgeOnNoConsumers, boolean autoCreateAddress) throws Exception; + Queue createQueue(SimpleString address, RoutingType routingType, SimpleString queueName, SimpleString filter, + SimpleString user, boolean durable, boolean temporary, boolean ignoreIfExists, boolean transientQueue, + boolean autoCreated, int maxConsumers, boolean purgeOnNoConsumers, boolean exclusive, boolean lastValue, boolean autoCreateAddress) throws Exception; + @Deprecated Queue createQueue(SimpleString address, SimpleString queueName, SimpleString filter, boolean durable, boolean temporary) throws Exception; @@ -426,6 +437,12 @@ public interface ActiveMQServer extends ServiceComponent { Integer maxConsumers, Boolean purgeOnNoConsumers) throws Exception; + Queue updateQueue(String name, + RoutingType routingType, + Integer maxConsumers, + Boolean purgeOnNoConsumers, + Boolean exclusive) throws Exception; + /* * add a ProtocolManagerFactory to be used. Note if @see Configuration#isResolveProtocols is tur then this factory will * replace any factories with the same protocol http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/BindingQueryResult.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/BindingQueryResult.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/BindingQueryResult.java index 7f340b1..a76812c 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/BindingQueryResult.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/BindingQueryResult.java @@ -37,13 +37,19 @@ public class BindingQueryResult { private final AddressInfo addressInfo; + private boolean defaultExclusive; + + private boolean defaultLastValue; + public BindingQueryResult(final boolean exists, final AddressInfo addressInfo, final List<SimpleString> queueNames, final boolean autoCreateQueues, final boolean autoCreateAddresses, final boolean defaultPurgeOnNoConsumers, - final int defaultMaxConsumers) { + final int defaultMaxConsumers, + final boolean defaultExclusive, + final boolean defaultLastValue) { this.addressInfo = addressInfo; this.exists = exists; @@ -57,6 +63,10 @@ public class BindingQueryResult { this.defaultPurgeOnNoConsumers = defaultPurgeOnNoConsumers; this.defaultMaxConsumers = defaultMaxConsumers; + + this.defaultExclusive = defaultExclusive; + + this.defaultLastValue = defaultLastValue; } public boolean isExists() { @@ -86,4 +96,12 @@ public class BindingQueryResult { public int getDefaultMaxConsumers() { return defaultMaxConsumers; } + + public boolean isDefaultExclusive() { + return defaultExclusive; + } + + public boolean isDefaultLastValue() { + return defaultLastValue; + } } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java index b39f4da..ff4e82b 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 @@ -68,6 +68,12 @@ public interface Queue extends Bindable,CriticalComponent { void setPurgeOnNoConsumers(boolean value); + boolean isExclusive(); + + void setExclusive(boolean value); + + boolean isLastValue(); + int getMaxConsumers(); void setMaxConsumer(int maxConsumers); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/QueueConfig.java ---------------------------------------------------------------------- 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 e879966..75f859d 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 @@ -37,6 +37,8 @@ public final class QueueConfig { private final boolean autoCreated; private final RoutingType routingType; private final int maxConsumers; + private final boolean exclusive; + private final boolean lastValue; private final boolean purgeOnNoConsumers; public static final class Builder { @@ -52,6 +54,8 @@ public final class QueueConfig { private boolean autoCreated; private RoutingType routingType; private int maxConsumers; + private boolean exclusive; + private boolean lastValue; private boolean purgeOnNoConsumers; private Builder(final long id, final SimpleString name) { @@ -70,6 +74,8 @@ public final class QueueConfig { this.autoCreated = true; this.routingType = ActiveMQDefaultConfiguration.getDefaultRoutingType(); this.maxConsumers = ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(); + this.exclusive = ActiveMQDefaultConfiguration.getDefaultExclusive(); + this.lastValue = ActiveMQDefaultConfiguration.getDefaultLastValue(); this.purgeOnNoConsumers = ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(); validateState(); } @@ -122,6 +128,17 @@ public final class QueueConfig { return this; } + public Builder exclusive(final boolean exclusive) { + this.exclusive = exclusive; + return this; + } + + public Builder lastValue(final boolean lastValue) { + this.lastValue = lastValue; + return this; + } + + public Builder purgeOnNoConsumers(final boolean purgeOnNoConsumers) { this.purgeOnNoConsumers = purgeOnNoConsumers; return this; @@ -153,7 +170,7 @@ public final class QueueConfig { } else { pageSubscription = null; } - return new QueueConfig(id, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, purgeOnNoConsumers); + return new QueueConfig(id, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, exclusive, lastValue, purgeOnNoConsumers); } } @@ -197,6 +214,8 @@ public final class QueueConfig { final boolean autoCreated, final RoutingType routingType, final int maxConsumers, + final boolean exclusive, + final boolean lastValue, final boolean purgeOnNoConsumers) { this.id = id; this.address = address; @@ -209,6 +228,8 @@ public final class QueueConfig { this.autoCreated = autoCreated; this.routingType = routingType; this.purgeOnNoConsumers = purgeOnNoConsumers; + this.exclusive = exclusive; + this.lastValue = lastValue; this.maxConsumers = maxConsumers; } @@ -256,6 +277,14 @@ public final class QueueConfig { return maxConsumers; } + public boolean isExclusive() { + return exclusive; + } + + public boolean isLastValue() { + return lastValue; + } + public RoutingType deliveryMode() { return routingType; } @@ -289,6 +318,10 @@ public final class QueueConfig { return false; if (maxConsumers != that.maxConsumers) return false; + if (exclusive != that.exclusive) + return false; + if (lastValue != that.lastValue) + return false; if (purgeOnNoConsumers != that.purgeOnNoConsumers) return false; return user != null ? user.equals(that.user) : that.user == null; @@ -308,6 +341,8 @@ public final class QueueConfig { result = 31 * result + (autoCreated ? 1 : 0); result = 31 * result + routingType.getType(); result = 31 * result + maxConsumers; + result = 31 * result + (exclusive ? 1 : 0); + result = 31 * result + (lastValue ? 1 : 0); result = 31 * result + (purgeOnNoConsumers ? 1 : 0); return result; } @@ -326,6 +361,8 @@ public final class QueueConfig { + ", autoCreated=" + autoCreated + ", routingType=" + routingType + ", maxConsumers=" + maxConsumers + + ", exclusive=" + exclusive + + ", lastValue=" + lastValue + ", purgeOnNoConsumers=" + purgeOnNoConsumers + '}'; } } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerSession.java ---------------------------------------------------------------------- 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 d7e1ad2..59a400b 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 @@ -155,6 +155,18 @@ public interface ServerSession extends SecurityAuth { SimpleString filterString, boolean temporary, boolean durable, + int maxConsumers, + boolean purgeOnNoConsumers, + Boolean exclusive, + Boolean lastValue, + boolean autoCreated) throws Exception; + + Queue createQueue(SimpleString address, + SimpleString name, + RoutingType routingType, + SimpleString filterString, + boolean temporary, + boolean durable, boolean autoCreated) throws Exception; Queue createQueue(AddressInfo addressInfo, @@ -164,6 +176,15 @@ public interface ServerSession extends SecurityAuth { boolean durable, boolean autoCreated) throws Exception; + Queue createQueue(AddressInfo addressInfo, + SimpleString name, + SimpleString filterString, + boolean temporary, + boolean durable, + Boolean exclusive, + Boolean lastValue, + boolean autoCreated) throws Exception; + AddressInfo createAddress(SimpleString address, EnumSet<RoutingType> routingTypes, boolean autoCreated) throws Exception; @@ -254,6 +275,16 @@ public interface ServerSession extends SecurityAuth { boolean isClosed(); void createSharedQueue(SimpleString address, + SimpleString name, + RoutingType routingType, + SimpleString filterString, + boolean durable, + Integer maxConsumers, + Boolean purgeOnNoConsumers, + Boolean exclusive, + Boolean lastValue) throws Exception; + + void createSharedQueue(SimpleString address, SimpleString name, RoutingType routingType, boolean durable, http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java index 6ce681e..d7eb2b9 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 @@ -825,6 +825,8 @@ public class ActiveMQServerImpl implements ActiveMQServer { boolean autoCreateAddresses = addressSettings.isAutoCreateAddresses(); boolean defaultPurgeOnNoConsumers = addressSettings.isDefaultPurgeOnNoConsumers(); int defaultMaxConsumers = addressSettings.getDefaultMaxConsumers(); + boolean defaultExclusive = addressSettings.isDefaultExclusiveQueue(); + boolean defaultLastValie = addressSettings.isDefaultLastValueQueue(); List<SimpleString> names = new ArrayList<>(); @@ -833,7 +835,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { SimpleString bindAddress = new SimpleString(realAddress); if (managementService != null) { if (bindAddress.equals(managementService.getManagementAddress())) { - return new BindingQueryResult(true, null, names, autoCreateQeueus, autoCreateAddresses, defaultPurgeOnNoConsumers, defaultMaxConsumers); + return new BindingQueryResult(true, null, names, autoCreateQeueus, autoCreateAddresses, defaultPurgeOnNoConsumers, defaultMaxConsumers, defaultExclusive, defaultLastValie); } } @@ -851,7 +853,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { AddressInfo info = getAddressInfo(bindAddress); - return new BindingQueryResult(info != null, info, names, autoCreateQeueus, autoCreateAddresses, defaultPurgeOnNoConsumers, defaultMaxConsumers); + return new BindingQueryResult(info != null, info, names, autoCreateQeueus, autoCreateAddresses, defaultPurgeOnNoConsumers, defaultMaxConsumers, defaultExclusive, defaultLastValie); } @Override @@ -863,6 +865,8 @@ public class ActiveMQServerImpl implements ActiveMQServer { boolean autoCreateQueues = getAddressSettingsRepository().getMatch(name.toString()).isAutoCreateQueues(); boolean defaultPurgeOnNoConsumers = getAddressSettingsRepository().getMatch(name.toString()).isDefaultPurgeOnNoConsumers(); int defaultMaxConsumers = getAddressSettingsRepository().getMatch(name.toString()).getDefaultMaxConsumers(); + boolean defaultExclusiveQueue = getAddressSettingsRepository().getMatch(name.toString()).isDefaultExclusiveQueue(); + boolean defaultLastValueQueue = getAddressSettingsRepository().getMatch(name.toString()).isDefaultLastValueQueue(); QueueQueryResult response; @@ -877,14 +881,14 @@ public class ActiveMQServerImpl implements ActiveMQServer { SimpleString filterString = filter == null ? null : filter.getFilterString(); - response = new QueueQueryResult(name, binding.getAddress(), queue.isDurable(), queue.isTemporary(), filterString, queue.getConsumerCount(), queue.getMessageCount(), autoCreateQueues, true, queue.isAutoCreated(), queue.isPurgeOnNoConsumers(), queue.getRoutingType(), queue.getMaxConsumers()); + response = new QueueQueryResult(name, binding.getAddress(), queue.isDurable(), queue.isTemporary(), filterString, queue.getConsumerCount(), queue.getMessageCount(), autoCreateQueues, true, queue.isAutoCreated(), queue.isPurgeOnNoConsumers(), queue.getRoutingType(), queue.getMaxConsumers(), queue.isExclusive(), queue.isLastValue()); } else if (name.equals(managementAddress)) { // make an exception for the management address (see HORNETQ-29) - response = new QueueQueryResult(name, managementAddress, true, false, null, -1, -1, autoCreateQueues, true, false, false, RoutingType.MULTICAST, -1); + response = new QueueQueryResult(name, managementAddress, true, false, null, -1, -1, autoCreateQueues, true, false, false, RoutingType.MULTICAST, -1, false, false); } else if (autoCreateQueues) { - response = new QueueQueryResult(name, name, true, false, null, 0, 0, true, false, false, defaultPurgeOnNoConsumers, RoutingType.MULTICAST, defaultMaxConsumers); + response = new QueueQueryResult(name, name, true, false, null, 0, 0, true, false, false, defaultPurgeOnNoConsumers, RoutingType.MULTICAST, defaultMaxConsumers, defaultExclusiveQueue, defaultLastValueQueue); } else { - response = new QueueQueryResult(null, null, false, false, null, 0, 0, false, false, false, false, RoutingType.MULTICAST, 0); + response = new QueueQueryResult(null, null, false, false, null, 0, 0, false, false, false, false, RoutingType.MULTICAST, 0, null, null); } return response; @@ -1672,9 +1676,25 @@ 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 { - return createQueue(addressInfo, queueName, filter, user, durable, temporary, false, false, autoCreated, maxConsumers, purgeOnNoConsumers, autoCreateAddress); + 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.isDefaultLastValueQueue(), autoCreateAddress); } + @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 { + return createQueue(addressInfo, queueName, filter, user, durable, temporary, false, false, autoCreated, maxConsumers, purgeOnNoConsumers, exclusive, lastValue, autoCreateAddress); + } + + + @Override + public Queue createQueue(SimpleString address, RoutingType routingType, SimpleString queueName, SimpleString filter, + 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.isDefaultLastValueQueue(), autoCreateAddress); + } + + @Deprecated @Override public Queue createQueue(final SimpleString address, @@ -1692,6 +1712,21 @@ public class ActiveMQServerImpl implements ActiveMQServer { final SimpleString filterString, final SimpleString user, boolean durable) throws Exception { + AddressSettings as = getAddressSettingsRepository().getMatch(address == null ? name.toString() : address.toString()); + createSharedQueue(address, routingType, name, filterString, user, durable, as.getDefaultMaxConsumers(), as.isDefaultPurgeOnNoConsumers(), as.isDefaultExclusiveQueue(), as.isDefaultLastValueQueue()); + } + + @Override + public void createSharedQueue(final SimpleString address, + RoutingType routingType, + final SimpleString name, + final SimpleString filterString, + final SimpleString user, + boolean durable, + int maxConsumers, + boolean purgeOnNoConsumers, + boolean exclusive, + boolean lastValue) throws Exception { //force the old contract about address if (address == null) { throw new NullPointerException("address can't be null!"); @@ -1705,7 +1740,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { } } - final Queue queue = createQueue(address, routingType, name, filterString, user, durable, !durable, true, !durable, false, Queue.MAX_CONSUMERS_UNLIMITED, false, true); + final Queue queue = createQueue(address, routingType, name, filterString, user, durable, !durable, true, !durable, false, maxConsumers, purgeOnNoConsumers, exclusive, lastValue, true); if (!queue.getAddress().equals(address)) { throw ActiveMQMessageBundle.BUNDLE.queueSubscriptionBelongsToDifferentAddress(name); @@ -2480,16 +2515,19 @@ public class ActiveMQServerImpl implements ActiveMQServer { for (CoreQueueConfiguration config : queues) { SimpleString queueName = SimpleString.toSimpleString(config.getName()); ActiveMQServerLogger.LOGGER.deployQueue(config.getName(), config.getAddress()); - + AddressSettings as = addressSettingsRepository.getMatch(config.getAddress()); // determine if there is an address::queue match; update it if so if (locateQueue(queueName) != null && locateQueue(queueName).getAddress().toString().equals(config.getAddress())) { - updateQueue(config.getName(), config.getRoutingType(), config.getMaxConsumers(), config.getPurgeOnNoConsumers()); + updateQueue(config.getName(), config.getRoutingType(), config.getMaxConsumers(), config.getPurgeOnNoConsumers(), + config.isExclusive() == null ? as.isDefaultExclusiveQueue() : config.isExclusive()); } else { // if the address::queue doesn't exist then create it try { createQueue(SimpleString.toSimpleString(config.getAddress()), config.getRoutingType(), queueName, SimpleString.toSimpleString(config.getFilterString()), SimpleString.toSimpleString(config.getUser()), - config.isDurable(),false,false,false,false,config.getMaxConsumers(),config.getPurgeOnNoConsumers(),true); + config.isDurable(),false,false,false,false,config.getMaxConsumers(),config.getPurgeOnNoConsumers(), + config.isExclusive() == null ? as.isDefaultExclusiveQueue() : config.isExclusive(), + config.isLastValue() == null ? as.isDefaultLastValueQueue() : config.isLastValue(), true); } catch (ActiveMQQueueExistsException e) { // the queue may exist on a *different* address ActiveMQServerLogger.LOGGER.warn(e.getMessage()); @@ -2668,6 +2706,8 @@ public class ActiveMQServerImpl implements ActiveMQServer { final boolean autoCreated, final int maxConsumers, final boolean purgeOnNoConsumers, + final boolean exclusive, + final boolean lastValue, final boolean autoCreateAddress) throws Exception { final QueueBinding binding = (QueueBinding) postOffice.getBinding(queueName); if (binding != null) { @@ -2710,7 +2750,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { throw ActiveMQMessageBundle.BUNDLE.invalidRoutingTypeForAddress(rt, info.getName().toString(), info.getRoutingTypes()); } - final QueueConfig queueConfig = queueConfigBuilder.filter(filter).pagingManager(pagingManager).user(user).durable(durable).temporary(temporary).autoCreated(autoCreated).routingType(addrInfo.getRoutingType()).maxConsumers(maxConsumers).purgeOnNoConsumers(purgeOnNoConsumers).build(); + final QueueConfig queueConfig = queueConfigBuilder.filter(filter).pagingManager(pagingManager).user(user).durable(durable).temporary(temporary).autoCreated(autoCreated).routingType(addrInfo.getRoutingType()).maxConsumers(maxConsumers).purgeOnNoConsumers(purgeOnNoConsumers).exclusive(exclusive).lastValue(lastValue).build(); callBrokerPlugins(hasBrokerPlugins() ? plugin -> plugin.beforeCreateQueue(queueConfig) : null); @@ -2776,6 +2816,8 @@ public class ActiveMQServerImpl implements ActiveMQServer { final boolean autoCreated, final int maxConsumers, final boolean purgeOnNoConsumers, + final boolean exclusive, + final boolean lastValue, final boolean autoCreateAddress) throws Exception { final QueueBinding binding = (QueueBinding) postOffice.getBinding(queueName); @@ -2817,7 +2859,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { throw ActiveMQMessageBundle.BUNDLE.invalidRoutingTypeForAddress(routingType, info.getName().toString(), info.getRoutingTypes()); } - final QueueConfig queueConfig = queueConfigBuilder.filter(filter).pagingManager(pagingManager).user(user).durable(durable).temporary(temporary).autoCreated(autoCreated).routingType(routingType).maxConsumers(maxConsumers).purgeOnNoConsumers(purgeOnNoConsumers).build(); + final QueueConfig queueConfig = queueConfigBuilder.filter(filter).pagingManager(pagingManager).user(user).durable(durable).temporary(temporary).autoCreated(autoCreated).routingType(routingType).maxConsumers(maxConsumers).purgeOnNoConsumers(purgeOnNoConsumers).exclusive(exclusive).lastValue(lastValue).build(); callBrokerPlugins(hasBrokerPlugins() ? plugin -> plugin.beforeCreateQueue(queueConfig) : null); @@ -2873,8 +2915,16 @@ public class ActiveMQServerImpl implements ActiveMQServer { RoutingType routingType, Integer maxConsumers, Boolean purgeOnNoConsumers) throws Exception { + return updateQueue(name, routingType, maxConsumers, purgeOnNoConsumers, null); + } - final QueueBinding queueBinding = this.postOffice.updateQueue(new SimpleString(name), routingType, maxConsumers, purgeOnNoConsumers); + @Override + public Queue updateQueue(String name, + RoutingType routingType, + Integer maxConsumers, + Boolean purgeOnNoConsumers, + Boolean exclusive) throws Exception { + final QueueBinding queueBinding = this.postOffice.updateQueue(new SimpleString(name), routingType, maxConsumers, purgeOnNoConsumers, exclusive); if (queueBinding != null) { final Queue queue = queueBinding.getQueue(); return queue; http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java ---------------------------------------------------------------------- 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 5951a01..7aada5e 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 @@ -60,6 +60,7 @@ public class LastValueQueue extends QueueImpl { final boolean autoCreated, final RoutingType routingType, final Integer maxConsumers, + final Boolean exclusive, final Boolean purgeOnNoConsumers, final ScheduledExecutorService scheduledExecutor, final PostOffice postOffice, @@ -68,7 +69,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, purgeOnNoConsumers, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server, factory); + super(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, routingType, maxConsumers, exclusive, purgeOnNoConsumers, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server, factory); } @Override @@ -162,6 +163,11 @@ public class LastValueQueue extends QueueImpl { super.refRemoved(ref); } + @Override + public boolean isLastValue() { + return true; + } + private class HolderReference implements MessageReference { private final SimpleString prop; http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java index 96b6662..cfd06d9 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 @@ -150,6 +150,7 @@ public class PostOfficeJournalLoader implements JournalLoader { .autoCreated(queueBindingInfo.isAutoCreated()) .purgeOnNoConsumers(queueBindingInfo.isPurgeOnNoConsumers()) .maxConsumers(queueBindingInfo.getMaxConsumers()) + .exclusive(queueBindingInfo.isExclusive()) .routingType(RoutingType.getType(queueBindingInfo.getRoutingType())); final Queue queue = queueFactory.createQueueWith(queueConfigBuilder.build()); queue.setConsumersRefCount(new QueueManagerImpl(((PostOfficeImpl)postOffice).getServer(), queueBindingInfo.getQueueName())); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java index 64da8b0..7f23d09 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 @@ -72,12 +72,11 @@ public class QueueFactoryImpl implements QueueFactory { @Override public Queue createQueueWith(final QueueConfig config) { - final AddressSettings addressSettings = addressSettingsRepository.getMatch(config.address().toString()); final Queue queue; - if (addressSettings.isLastValueQueue()) { - 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.isPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this); + if (config.isLastValue()) { + 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.isPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this); } 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.isPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this); + 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.isPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this); } server.getCriticalAnalyzer().add(queue); @@ -102,8 +101,8 @@ public class QueueFactoryImpl implements QueueFactory { AddressSettings addressSettings = addressSettingsRepository.getMatch(address.toString()); Queue queue; - if (addressSettings.isLastValueQueue()) { - queue = new LastValueQueue(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, ActiveMQDefaultConfiguration.getDefaultRoutingType(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this); + if (addressSettings.isDefaultLastValueQueue()) { + queue = new LastValueQueue(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, ActiveMQDefaultConfiguration.getDefaultRoutingType(), ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers(), ActiveMQDefaultConfiguration.getDefaultExclusive(), ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers(), scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this); } else { queue = new QueueImpl(persistenceID, address, name, filter, pageSubscription, user, durable, temporary, autoCreated, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executorFactory.getExecutor(), server, this); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47c9a90d/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java index c527c46..dbf79e2 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 @@ -257,6 +257,8 @@ public class QueueImpl extends CriticalComponentImpl implements Queue { private volatile int maxConsumers; + private volatile boolean exclusive; + private volatile boolean purgeOnNoConsumers; private final AddressInfo addressInfo; @@ -389,6 +391,29 @@ 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, null, purgeOnNoConsumers, scheduledExecutor, postOffice, storageManager, addressSettingsRepository, executor, server, factory); + } + + public QueueImpl(final long id, + final SimpleString address, + final SimpleString name, + final Filter filter, + final PageSubscription pageSubscription, + final SimpleString user, + final boolean durable, + final boolean temporary, + final boolean autoCreated, + final RoutingType routingType, + final Integer maxConsumers, + final Boolean exclusive, + final Boolean purgeOnNoConsumers, + final ScheduledExecutorService scheduledExecutor, + final PostOffice postOffice, + final StorageManager storageManager, + final HierarchicalRepository<AddressSettings> addressSettingsRepository, + final ArtemisExecutor executor, + final ActiveMQServer server, + final QueueFactory factory) { super(server == null ? EmptyCriticalAnalyzer.getInstance() : server.getCriticalAnalyzer(), CRITICAL_PATHS); this.id = id; @@ -413,6 +438,8 @@ public class QueueImpl extends CriticalComponentImpl implements Queue { this.maxConsumers = maxConsumers == null ? ActiveMQDefaultConfiguration.getDefaultMaxQueueConsumers() : maxConsumers; + this.exclusive = exclusive == null ? ActiveMQDefaultConfiguration.getDefaultExclusive() : exclusive; + this.purgeOnNoConsumers = purgeOnNoConsumers == null ? ActiveMQDefaultConfiguration.getDefaultPurgeOnNoConsumers() : purgeOnNoConsumers; this.postOffice = postOffice; @@ -463,7 +490,18 @@ public class QueueImpl extends CriticalComponentImpl implements Queue { return user; } + @Override public boolean isExclusive() { + return exclusive; + } + + @Override + public synchronized void setExclusive(boolean exclusive) { + this.exclusive = exclusive; + } + + @Override + public boolean isLastValue() { return false; } @@ -2190,6 +2228,10 @@ public class QueueImpl extends CriticalComponentImpl implements Queue { } } + if (exclusive) { + consumer = consumerList.get(0).consumer; + } + HandleStatus status = handle(ref, consumer); if (status == HandleStatus.HANDLED) { @@ -2237,7 +2279,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue { // Only move onto the next position if the consumer on the current position was used. // When using group we don't need to load balance to the next position - if (groupConsumer == null) { + if (!exclusive && groupConsumer == null) { pos++; } @@ -2730,8 +2772,12 @@ public class QueueImpl extends CriticalComponentImpl implements Queue { } } + if (exclusive) { + consumer = consumerList.get(0).consumer; + } + // Only move onto the next position if the consumer on the current position was used. - if (groupConsumer == null) { + if (!exclusive && groupConsumer == null) { pos++; }
