You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2014/02/17 21:19:38 UTC
svn commit: r1569102 [2/5] - in
/qpid/branches/java-broker-amqp-1-0-management/java:
amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/
bdbstore/src/test/java/org/ap...
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java Mon Feb 17 20:19:36 2014
@@ -20,12 +20,14 @@
*/
package org.apache.qpid.server.queue;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.apache.qpid.server.exchange.AMQUnknownExchangeType;
+import org.apache.qpid.server.model.ExclusivityPolicy;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.server.configuration.BrokerProperties;
@@ -36,6 +38,7 @@ import org.apache.qpid.server.model.Queu
import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.store.DurableConfigurationStoreHelper;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
+import org.apache.qpid.server.util.MapValueConverter;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.server.virtualhost.ExchangeExistsException;
import org.apache.qpid.server.virtualhost.ReservedExchangeNameException;
@@ -45,7 +48,6 @@ import org.apache.qpid.server.virtualhos
public class AMQQueueFactory implements QueueFactory
{
- public static final String QPID_DEFAULT_LVQ_KEY = "qpid.LVQ_key";
public static final String DEFAULT_DLQ_NAME_SUFFIX = "_DLQ";
@@ -59,381 +61,207 @@ public class AMQQueueFactory implements
{
_virtualHost = virtualHost;
_queueRegistry = queueRegistry;
+ }
+
+ @Override
+ public AMQQueue restoreQueue(Map<String, Object> attributes) throws QpidSecurityException
+ {
+ return createOrRestoreQueue(null, attributes, false);
+
}
- private abstract static class QueueProperty
+ @Override
+ public AMQQueue createQueue(final AMQSessionModel creatingSession,
+ Map<String, Object> attributes) throws QpidSecurityException
+ {
+ return createOrRestoreQueue(creatingSession, attributes, true);
+ }
+
+ private AMQQueue createOrRestoreQueue(final AMQSessionModel creatingSession, Map<String, Object> attributes,
+ boolean createInStore) throws QpidSecurityException
{
- private final String _argumentName;
+ String queueName = MapValueConverter.getStringAttribute(Queue.NAME,attributes);
- public QueueProperty(String argumentName)
+ QueueConfiguration config = _virtualHost.getConfiguration().getQueueConfiguration(queueName);
+
+ if (!attributes.containsKey(Queue.ALERT_THRESHOLD_MESSAGE_AGE) && config.getMaximumMessageAge() != 0)
{
- _argumentName = argumentName;
+ attributes.put(Queue.ALERT_THRESHOLD_MESSAGE_AGE, config.getMaximumMessageAge());
}
-
- public String getArgumentName()
+ if (!attributes.containsKey(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_BYTES) && config.getMaximumQueueDepth() != 0)
{
- return _argumentName;
+ attributes.put(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_BYTES, config.getMaximumQueueDepth());
}
-
-
- public abstract void setPropertyValue(AMQQueue queue, Object value);
-
- }
-
- private abstract static class QueueLongProperty extends QueueProperty
- {
-
- public QueueLongProperty(String argumentName)
+ if (!attributes.containsKey(Queue.ALERT_THRESHOLD_MESSAGE_SIZE) && config.getMaximumMessageSize() != 0)
{
- super(argumentName);
+ attributes.put(Queue.ALERT_THRESHOLD_MESSAGE_SIZE, config.getMaximumMessageSize());
}
-
- public void setPropertyValue(AMQQueue queue, Object value)
+ if (!attributes.containsKey(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES) && config.getMaximumMessageCount() != 0)
{
- if(value instanceof Number)
- {
- setPropertyValue(queue, ((Number)value).longValue());
- }
-
+ attributes.put(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES, config.getMaximumMessageCount());
}
-
- abstract void setPropertyValue(AMQQueue queue, long value);
-
-
- }
-
- private abstract static class QueueIntegerProperty extends QueueProperty
- {
- public QueueIntegerProperty(String argumentName)
+ if (!attributes.containsKey(Queue.ALERT_REPEAT_GAP) && config.getMinimumAlertRepeatGap() != 0)
{
- super(argumentName);
+ attributes.put(Queue.ALERT_REPEAT_GAP, config.getMinimumAlertRepeatGap());
}
-
- public void setPropertyValue(AMQQueue queue, Object value)
+ if (config.getMaxDeliveryCount() != 0 && !attributes.containsKey(Queue.MAXIMUM_DELIVERY_ATTEMPTS))
{
- if(value instanceof Number)
- {
- setPropertyValue(queue, ((Number)value).intValue());
- }
-
+ attributes.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, config.getMaxDeliveryCount());
}
- abstract void setPropertyValue(AMQQueue queue, int value);
- }
-
- private static final QueueProperty[] DECLARABLE_PROPERTIES = {
- new QueueLongProperty(Queue.ALERT_THRESHOLD_MESSAGE_AGE)
- {
- public void setPropertyValue(AMQQueue queue, long value)
- {
- queue.setMaximumMessageAge(value);
- }
- },
- new QueueLongProperty(Queue.ALERT_THRESHOLD_MESSAGE_SIZE)
- {
- public void setPropertyValue(AMQQueue queue, long value)
- {
- queue.setMaximumMessageSize(value);
- }
- },
- new QueueLongProperty(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES)
- {
- public void setPropertyValue(AMQQueue queue, long value)
- {
- queue.setMaximumMessageCount(value);
- }
- },
- new QueueLongProperty(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_BYTES)
- {
- public void setPropertyValue(AMQQueue queue, long value)
- {
- queue.setMaximumQueueDepth(value);
- }
- },
- new QueueLongProperty(Queue.ALERT_REPEAT_GAP)
- {
- public void setPropertyValue(AMQQueue queue, long value)
- {
- queue.setMinimumAlertRepeatGap(value);
- }
- },
- new QueueLongProperty(Queue.QUEUE_FLOW_CONTROL_SIZE_BYTES)
- {
- public void setPropertyValue(AMQQueue queue, long value)
- {
- queue.setCapacity(value);
- }
- },
- new QueueLongProperty(Queue.QUEUE_FLOW_RESUME_SIZE_BYTES)
- {
- public void setPropertyValue(AMQQueue queue, long value)
- {
- queue.setFlowResumeCapacity(value);
- }
- },
- new QueueIntegerProperty(Queue.MAXIMUM_DELIVERY_ATTEMPTS)
- {
- public void setPropertyValue(AMQQueue queue, int value)
- {
- queue.setMaximumDeliveryCount(value);
- }
- }
- };
-
- @Override
- public AMQQueue restoreQueue(UUID id,
- String queueName,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- boolean deleteOnNoConsumer,
- Map<String, Object> arguments) throws QpidSecurityException
- {
- return createOrRestoreQueue(id, queueName, true, owner, autoDelete, exclusive, deleteOnNoConsumer, arguments, false);
-
- }
-
- /**
- * @param id the id to use.
- * @param deleteOnNoConsumer
- */
- @Override
- public AMQQueue createQueue(UUID id,
- String queueName,
- boolean durable,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- boolean deleteOnNoConsumer,
- Map<String, Object> arguments) throws QpidSecurityException
- {
- return createOrRestoreQueue(id, queueName, durable, owner, autoDelete, exclusive, deleteOnNoConsumer, arguments, true);
- }
-
- private AMQQueue createOrRestoreQueue(UUID id,
- String queueName,
- boolean durable,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- boolean deleteOnNoConsumer,
- Map<String, Object> arguments,
- boolean createInStore) throws QpidSecurityException
- {
- if (id == null)
+ if (!attributes.containsKey(Queue.QUEUE_FLOW_CONTROL_SIZE_BYTES) && config.getCapacity() != 0)
{
- throw new IllegalArgumentException("Queue id must not be null");
+ attributes.put(Queue.QUEUE_FLOW_CONTROL_SIZE_BYTES, config.getCapacity());
}
- if (queueName == null)
+ if (!attributes.containsKey(Queue.QUEUE_FLOW_RESUME_SIZE_BYTES) && config.getFlowResumeCapacity() != 0)
{
- throw new IllegalArgumentException("Queue name must not be null");
+ attributes.put(Queue.QUEUE_FLOW_RESUME_SIZE_BYTES, config.getFlowResumeCapacity());
}
- QueueConfiguration queueConfiguration = _virtualHost.getConfiguration().getQueueConfiguration(queueName);
-
- boolean createDLQ = createDLQ(autoDelete, arguments, queueConfiguration);
+ boolean createDLQ = createDLQ(attributes, config);
if (createDLQ)
{
validateDLNames(queueName);
}
- int priorities = 1;
- String conflationKey = null;
- String sortingKey = null;
-
- if(arguments != null)
- {
- if(arguments.containsKey(Queue.LVQ_KEY))
- {
- conflationKey = (String) arguments.get(Queue.LVQ_KEY);
- if(conflationKey == null)
- {
- conflationKey = QPID_DEFAULT_LVQ_KEY;
- }
- }
- else if(arguments.containsKey(Queue.PRIORITIES))
- {
- Object prioritiesObj = arguments.get(Queue.PRIORITIES);
- if(prioritiesObj instanceof Number)
- {
- priorities = ((Number)prioritiesObj).intValue();
- }
- else if(prioritiesObj instanceof String)
- {
- try
- {
- priorities = Integer.parseInt(prioritiesObj.toString());
- }
- catch (NumberFormatException e)
- {
- // TODO - should warn here of invalid format
- }
- }
- else
- {
- // TODO - should warn here of invalid format
- }
- }
- else if(arguments.containsKey(Queue.SORT_KEY))
- {
- sortingKey = (String)arguments.get(Queue.SORT_KEY);
- }
- }
+ AMQQueue queue;
- AMQQueue q;
- if(sortingKey != null)
+ if(attributes.containsKey(Queue.SORT_KEY))
{
- q = new SortedQueue(id, queueName, durable, owner, autoDelete, exclusive, _virtualHost, arguments, sortingKey);
+ queue = new SortedQueue(_virtualHost, creatingSession, attributes);
}
- else if(conflationKey != null)
+ else if(attributes.containsKey(Queue.LVQ_KEY))
{
- q = new ConflationQueue(id, queueName, durable, owner, autoDelete, exclusive, _virtualHost, arguments, conflationKey);
+ queue = new ConflationQueue(_virtualHost, creatingSession, attributes);
}
- else if(priorities > 1)
+ else if(attributes.containsKey(Queue.PRIORITIES))
{
- q = new PriorityQueue(id, queueName, durable, owner, autoDelete, exclusive, _virtualHost, arguments, priorities);
+ queue = new PriorityQueue(_virtualHost, creatingSession, attributes);
}
else
{
- q = new StandardQueue(id, queueName, durable, owner, autoDelete, exclusive, _virtualHost, arguments);
+ queue = new StandardQueue(_virtualHost, creatingSession, attributes);
}
- q.setDeleteOnNoConsumers(deleteOnNoConsumer);
-
//Register the new queue
- _queueRegistry.registerQueue(q);
+ _queueRegistry.registerQueue(queue);
- q.configure(_virtualHost.getConfiguration().getQueueConfiguration(queueName));
-
- if(arguments != null)
+ if(createDLQ)
{
- for(QueueProperty p : DECLARABLE_PROPERTIES)
- {
- if(arguments.containsKey(p.getArgumentName()))
- {
- p.setPropertyValue(q, arguments.get(p.getArgumentName()));
- }
- }
-
- if(arguments.get(Queue.NO_LOCAL) instanceof Boolean)
- {
- q.setNoLocal((Boolean)arguments.get(Queue.NO_LOCAL));
- }
-
+ createDLQ(queue);
}
-
- if(createDLQ)
+ else if(attributes != null && attributes.get(Queue.ALTERNATE_EXCHANGE) instanceof String)
{
- final String dlExchangeName = getDeadLetterExchangeName(queueName);
- final String dlQueueName = getDeadLetterQueueName(queueName);
-
- Exchange dlExchange = null;
- final UUID dlExchangeId = UUIDGenerator.generateExchangeUUID(dlExchangeName, _virtualHost.getName());
+ final String altExchangeAttr = (String) attributes.get(Queue.ALTERNATE_EXCHANGE);
+ Exchange altExchange;
try
{
- dlExchange = _virtualHost.createExchange(dlExchangeId,
- dlExchangeName,
- ExchangeDefaults.FANOUT_EXCHANGE_CLASS,
- true, false, null);
- }
- catch(ExchangeExistsException e)
- {
- // We're ok if the exchange already exists
- dlExchange = e.getExistingExchange();
- }
- catch (ReservedExchangeNameException e)
- {
- throw new ConnectionScopedRuntimeException("Attempt to create an alternate exchange for a queue failed",e);
- }
- catch (AMQUnknownExchangeType e)
- {
- throw new ConnectionScopedRuntimeException("Attempt to create an alternate exchange for a queue failed",e);
+ altExchange = _virtualHost.getExchange(UUID.fromString(altExchangeAttr));
}
- catch (UnknownExchangeException e)
+ catch(IllegalArgumentException e)
{
- throw new ConnectionScopedRuntimeException("Attempt to create an alternate exchange for a queue failed",e);
+ altExchange = _virtualHost.getExchange(altExchangeAttr);
}
+ queue.setAlternateExchange(altExchange);
+ }
- AMQQueue dlQueue = null;
+ if (createInStore && queue.isDurable() && !(queue.getLifetimePolicy()
+ == LifetimePolicy.DELETE_ON_CONNECTION_CLOSE
+ || queue.getLifetimePolicy()
+ == LifetimePolicy.DELETE_ON_SESSION_END))
+ {
+ DurableConfigurationStoreHelper.createQueue(_virtualHost.getDurableConfigurationStore(), queue);
+ }
- synchronized(_queueRegistry)
- {
- dlQueue = _queueRegistry.getQueue(dlQueueName);
+ return queue;
+ }
- if(dlQueue == null)
- {
- //set args to disable DLQ-ing/MDC from the DLQ itself, preventing loops etc
- final Map<String, Object> args = new HashMap<String, Object>();
- args.put(Queue.CREATE_DLQ_ON_CREATION, false);
- args.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, 0);
-
- try
- {
- dlQueue = _virtualHost.createQueue(UUIDGenerator.generateQueueUUID(dlQueueName, _virtualHost.getName()), dlQueueName, true, owner, false, exclusive,
- false, args);
- }
- catch (QueueExistsException e)
- {
- throw new ServerScopedRuntimeException("Attempt to create a queue failed because the " +
- "queue already exists, however this occurred within " +
- "a block where the queue existence had previously been " +
- "checked, and no queue creation should have been " +
- "possible from another thread", e);
- }
- }
- }
+ private void createDLQ(final AMQQueue queue) throws QpidSecurityException
+ {
+ final String queueName = queue.getName();
+ final String dlExchangeName = getDeadLetterExchangeName(queueName);
+ final String dlQueueName = getDeadLetterQueueName(queueName);
- //ensure the queue is bound to the exchange
- if(!dlExchange.isBound(DLQ_ROUTING_KEY, dlQueue))
- {
- //actual routing key used does not matter due to use of fanout exchange,
- //but we will make the key 'dlq' as it can be logged at creation.
- dlExchange.addBinding(DLQ_ROUTING_KEY, dlQueue, null);
- }
- q.setAlternateExchange(dlExchange);
+ Exchange dlExchange = null;
+ final UUID dlExchangeId = UUIDGenerator.generateExchangeUUID(dlExchangeName, _virtualHost.getName());
+
+ try
+ {
+ dlExchange = _virtualHost.createExchange(dlExchangeId,
+ dlExchangeName,
+ ExchangeDefaults.FANOUT_EXCHANGE_CLASS,
+ true, false, null);
+ }
+ catch(ExchangeExistsException e)
+ {
+ // We're ok if the exchange already exists
+ dlExchange = e.getExistingExchange();
+ }
+ catch (ReservedExchangeNameException e)
+ {
+ throw new ConnectionScopedRuntimeException("Attempt to create an alternate exchange for a queue failed",e);
}
- else if(arguments != null && arguments.get(Queue.ALTERNATE_EXCHANGE) instanceof String)
+ catch (AMQUnknownExchangeType e)
{
+ throw new ConnectionScopedRuntimeException("Attempt to create an alternate exchange for a queue failed",e);
+ }
+ catch (UnknownExchangeException e)
+ {
+ throw new ConnectionScopedRuntimeException("Attempt to create an alternate exchange for a queue failed",e);
+ }
- final String altExchangeAttr = (String) arguments.get(Queue.ALTERNATE_EXCHANGE);
- Exchange altExchange;
- try
- {
- altExchange = _virtualHost.getExchange(UUID.fromString(altExchangeAttr));
- }
- catch(IllegalArgumentException e)
+ AMQQueue dlQueue = null;
+
+ synchronized(_queueRegistry)
+ {
+ dlQueue = _queueRegistry.getQueue(dlQueueName);
+
+ if(dlQueue == null)
{
- altExchange = _virtualHost.getExchange(altExchangeAttr);
+ //set args to disable DLQ-ing/MDC from the DLQ itself, preventing loops etc
+ final Map<String, Object> args = new HashMap<String, Object>();
+ args.put(Queue.CREATE_DLQ_ON_CREATION, false);
+ args.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, 0);
+
+ try
+ {
+
+
+ args.put(Queue.ID, UUIDGenerator.generateQueueUUID(dlQueueName, _virtualHost.getName()));
+ args.put(Queue.NAME, dlQueueName);
+ args.put(Queue.DURABLE, true);
+ dlQueue = _virtualHost.createQueue(null, args);
+ }
+ catch (QueueExistsException e)
+ {
+ throw new ServerScopedRuntimeException("Attempt to create a queue failed because the " +
+ "queue already exists, however this occurred within " +
+ "a block where the queue existence had previously been " +
+ "checked, and no queue creation should have been " +
+ "possible from another thread", e);
+ }
}
- q.setAlternateExchange(altExchange);
}
- if (createInStore && q.isDurable() && !q.isAutoDelete())
+ //ensure the queue is bound to the exchange
+ if(!dlExchange.isBound(DLQ_ROUTING_KEY, dlQueue))
{
- DurableConfigurationStoreHelper.createQueue(_virtualHost.getDurableConfigurationStore(), q);
+ //actual routing key used does not matter due to use of fanout exchange,
+ //but we will make the key 'dlq' as it can be logged at creation.
+ dlExchange.addBinding(DLQ_ROUTING_KEY, dlQueue, null);
}
-
- return q;
+ queue.setAlternateExchange(dlExchange);
}
public AMQQueue createAMQQueueImpl(QueueConfiguration config) throws QpidSecurityException
{
- String queueName = config.getName();
-
- boolean durable = config.getDurable();
- boolean autodelete = config.getAutoDelete();
- boolean exclusive = config.getExclusive();
- String owner = config.getOwner();
- Map<String, Object> arguments = createQueueArgumentsFromConfig(config);
-
- // we need queues that are defined in config to have deterministic ids.
- UUID id = UUIDGenerator.generateQueueUUID(queueName, _virtualHost.getName());
- AMQQueue q = createQueue(id, queueName, durable, owner, autodelete, exclusive, false, arguments);
- q.configure(config);
+ Map<String, Object> arguments = createQueueAttributesFromConfig(_virtualHost, config);
+
+ AMQQueue q = createOrRestoreQueue(null, arguments, false);
return q;
}
@@ -471,16 +299,19 @@ public class AMQQueueFactory implements
/**
* Checks if DLQ is enabled for the queue.
*
- * @param autoDelete
- * queue auto-delete flag
* @param arguments
* queue arguments
* @param qConfig
* queue configuration
* @return true if DLQ enabled
*/
- protected static boolean createDLQ(boolean autoDelete, Map<String, Object> arguments, QueueConfiguration qConfig)
+ protected static boolean createDLQ(Map<String, Object> arguments, QueueConfiguration qConfig)
{
+ boolean autoDelete = MapValueConverter.getEnumAttribute(LifetimePolicy.class,
+ Queue.LIFETIME_POLICY,
+ arguments,
+ LifetimePolicy.PERMANENT) != LifetimePolicy.PERMANENT;
+
//feature is not to be enabled for temporary queues or when explicitly disabled by argument
if (!(autoDelete || (arguments != null && arguments.containsKey(Queue.ALTERNATE_EXCHANGE))))
{
@@ -525,46 +356,59 @@ public class AMQQueueFactory implements
return name + System.getProperty(BrokerProperties.PROPERTY_DEAD_LETTER_EXCHANGE_SUFFIX, DefaultExchangeFactory.DEFAULT_DLE_NAME_SUFFIX);
}
- private static Map<String, Object> createQueueArgumentsFromConfig(QueueConfiguration config)
+ private static Map<String, Object> createQueueAttributesFromConfig(final VirtualHost virtualHost,
+ QueueConfiguration config)
{
- Map<String,Object> arguments = new HashMap<String,Object>();
+ Map<String,Object> attributes = new HashMap<String,Object>();
if(config.getArguments() != null && !config.getArguments().isEmpty())
{
- arguments.putAll(QueueArgumentsConverter.convertWireArgsToModel(new HashMap<String, Object>(config.getArguments())));
+ attributes.putAll(QueueArgumentsConverter.convertWireArgsToModel(new HashMap<String, Object>(config.getArguments())));
}
if(config.isLVQ() || config.getLVQKey() != null)
{
- arguments.put(Queue.LVQ_KEY, config.getLVQKey() == null ? QPID_DEFAULT_LVQ_KEY : config.getLVQKey());
+ attributes.put(Queue.LVQ_KEY,
+ config.getLVQKey() == null ? ConflationQueue.DEFAULT_LVQ_KEY : config.getLVQKey());
}
else if (config.getPriority() || config.getPriorities() > 0)
{
- arguments.put(Queue.PRIORITIES, config.getPriorities() < 0 ? 10 : config.getPriorities());
+ attributes.put(Queue.PRIORITIES, config.getPriorities() < 0 ? 10 : config.getPriorities());
}
else if (config.getQueueSortKey() != null && !"".equals(config.getQueueSortKey()))
{
- arguments.put(Queue.SORT_KEY, config.getQueueSortKey());
+ attributes.put(Queue.SORT_KEY, config.getQueueSortKey());
}
if (!config.getAutoDelete() && config.isDeadLetterQueueEnabled())
{
- arguments.put(Queue.CREATE_DLQ_ON_CREATION, true);
+ attributes.put(Queue.CREATE_DLQ_ON_CREATION, true);
}
if (config.getDescription() != null && !"".equals(config.getDescription()))
{
- arguments.put(Queue.DESCRIPTION, config.getDescription());
+ attributes.put(Queue.DESCRIPTION, config.getDescription());
}
- if (arguments.isEmpty())
+ attributes.put(Queue.DURABLE, config.getDurable());
+ attributes.put(Queue.LIFETIME_POLICY,
+ config.getAutoDelete() ? LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS : LifetimePolicy.PERMANENT);
+ if(config.getExclusive())
{
- return Collections.emptyMap();
+ attributes.put(Queue.EXCLUSIVE, ExclusivityPolicy.CONTAINER);
}
- else
+ if(config.getOwner() != null)
{
- return arguments;
+ attributes.put(Queue.OWNER, config.getOwner());
}
+
+ attributes.put(Queue.NAME, config.getName());
+
+ // we need queues that are defined in config to have deterministic ids.
+ attributes.put(Queue.ID, UUIDGenerator.generateQueueUUID(config.getName(), virtualHost.getName()));
+
+
+ return attributes;
}
}
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueue.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueue.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueue.java Mon Feb 17 20:19:36 2014
@@ -22,22 +22,32 @@
package org.apache.qpid.server.queue;
import java.util.Map;
-import java.util.UUID;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.util.MapValueConverter;
import org.apache.qpid.server.virtualhost.VirtualHost;
public class ConflationQueue extends SimpleAMQQueue<ConflationQueueList.ConflationQueueEntry, ConflationQueue, ConflationQueueList>
{
- protected ConflationQueue(UUID id,
- String name,
- boolean durable,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- VirtualHost virtualHost,
- Map<String, Object> args, String conflationKey)
+ public static final String DEFAULT_LVQ_KEY = "qpid.LVQ_key";
+
+
+ protected ConflationQueue(VirtualHost virtualHost,
+ final AMQSessionModel creatingSession, Map<String, Object> attributes)
{
- super(id, name, durable, owner, autoDelete, exclusive, virtualHost, new ConflationQueueList.Factory(conflationKey), args);
+ super(virtualHost, creatingSession, attributes, entryList(attributes));
+ }
+
+ private static ConflationQueueList.Factory entryList(final Map<String, Object> attributes)
+ {
+
+ String conflationKey = MapValueConverter.getStringAttribute(Queue.LVQ_KEY,
+ attributes,
+ DEFAULT_LVQ_KEY);
+
+ // conflation key can still be null if it was present in the map with a null value
+ return new ConflationQueueList.Factory(conflationKey == null ? DEFAULT_LVQ_KEY : conflationKey);
}
public String getConflationKey()
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java Mon Feb 17 20:19:36 2014
@@ -20,19 +20,20 @@
*/
package org.apache.qpid.server.queue;
+import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Map;
-import java.util.UUID;
public abstract class OutOfOrderQueue<E extends QueueEntryImpl<E,Q,L>, Q extends OutOfOrderQueue<E,Q,L>, L extends SimpleQueueEntryList<E,Q,L>> extends SimpleAMQQueue<E,Q,L>
{
- protected OutOfOrderQueue(UUID id, String name, boolean durable,
- String owner, boolean autoDelete, boolean exclusive,
- VirtualHost virtualHost, QueueEntryListFactory<E,Q,L> entryListFactory, Map<String, Object> arguments)
+ protected OutOfOrderQueue(VirtualHost virtualHost,
+ final AMQSessionModel creatingSession,
+ Map<String, Object> attributes,
+ QueueEntryListFactory<E, Q, L> entryListFactory)
{
- super(id, name, durable, owner, autoDelete, exclusive, virtualHost, entryListFactory, arguments);
+ super(virtualHost, creatingSession, attributes, entryListFactory);
}
@Override
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueue.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueue.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueue.java Mon Feb 17 20:19:36 2014
@@ -20,23 +20,31 @@
*/
package org.apache.qpid.server.queue;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.util.MapValueConverter;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Map;
-import java.util.UUID;
public class PriorityQueue extends OutOfOrderQueue<PriorityQueueList.PriorityQueueEntry, PriorityQueue, PriorityQueueList>
{
- protected PriorityQueue(UUID id,
- final String name,
- final boolean durable,
- final String owner,
- final boolean autoDelete,
- boolean exclusive,
- final VirtualHost virtualHost,
- Map<String, Object> arguments, int priorities)
+
+ public static final int DEFAULT_PRIORITY_LEVELS = 10;
+
+ protected PriorityQueue(VirtualHost virtualHost,
+ final AMQSessionModel creatingSession,
+ Map<String, Object> attributes)
+ {
+ super(virtualHost, creatingSession, attributes, entryList(attributes));
+ }
+
+ private static PriorityQueueList.Factory entryList(final Map<String, Object> attributes)
{
- super(id, name, durable, owner, autoDelete, exclusive, virtualHost, new PriorityQueueList.Factory(priorities), arguments);
+ final Integer priorities = MapValueConverter.getIntegerAttribute(Queue.PRIORITIES, attributes,
+ DEFAULT_PRIORITY_LEVELS);
+
+ return new PriorityQueueList.Factory(priorities);
}
public int getPriorities()
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueArgumentsConverter.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueArgumentsConverter.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueArgumentsConverter.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueArgumentsConverter.java Mon Feb 17 20:19:36 2014
@@ -109,7 +109,7 @@ public class QueueArgumentsConverter
}
if(wireArguments.containsKey(QPID_LAST_VALUE_QUEUE) && !wireArguments.containsKey(QPID_LAST_VALUE_QUEUE_KEY))
{
- modelArguments.put(Queue.LVQ_KEY, AMQQueueFactory.QPID_DEFAULT_LVQ_KEY);
+ modelArguments.put(Queue.LVQ_KEY, ConflationQueue.DEFAULT_LVQ_KEY);
}
if(wireArguments.containsKey(QPID_SHARED_MSG_GROUP))
{
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueFactory.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueFactory.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueFactory.java Mon Feb 17 20:19:36 2014
@@ -22,25 +22,15 @@ package org.apache.qpid.server.queue;
import java.util.Map;
import java.util.UUID;
+
+import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.security.QpidSecurityException;
public interface QueueFactory
{
- AMQQueue createQueue(UUID id,
- String queueName,
- boolean durable,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- boolean deleteOnNoConsumer,
+ AMQQueue createQueue(final AMQSessionModel creatingSession,
Map<String, Object> arguments) throws QpidSecurityException;
- AMQQueue restoreQueue(UUID id,
- String queueName,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- boolean deleteOnNoConsumer,
- Map<String, Object> arguments) throws QpidSecurityException;
+ AMQQueue restoreQueue(Map<String, Object> arguments) throws QpidSecurityException;
}
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java Mon Feb 17 20:19:36 2014
@@ -18,6 +18,7 @@
*/
package org.apache.qpid.server.queue;
+import java.security.Principal;
import java.util.*;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -28,11 +29,14 @@ import java.util.concurrent.atomic.Atomi
import java.util.concurrent.atomic.AtomicLong;
import org.apache.log4j.Logger;
+import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.model.ExclusivityPolicy;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.pool.ReferenceCountingExecutorService;
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.configuration.BrokerProperties;
-import org.apache.qpid.server.configuration.QueueConfiguration;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.filter.FilterManager;
import org.apache.qpid.server.logging.LogActor;
@@ -50,12 +54,16 @@ import org.apache.qpid.server.protocol.A
import org.apache.qpid.server.security.AuthorizationHolder;
import org.apache.qpid.server.consumer.Consumer;
import org.apache.qpid.server.consumer.ConsumerTarget;
+import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.txn.AutoCommitTransaction;
import org.apache.qpid.server.txn.LocalTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
+import org.apache.qpid.server.util.Deletable;
+import org.apache.qpid.server.util.MapValueConverter;
+import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.server.util.StateChangeListener;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -78,19 +86,10 @@ abstract class SimpleAMQQueue<E extends
private final String _name;
/** null means shared */
- private final String _owner;
-
- private AuthorizationHolder _authorizationHolder;
-
- private boolean _exclusive = false;
- private AMQSessionModel _exclusiveOwner;
-
+ private String _description;
private final boolean _durable;
- /** If true, this queue is deleted when the last subscriber is removed */
- private final boolean _autoDelete;
-
private Exchange _alternateExchange;
@@ -142,6 +141,10 @@ abstract class SimpleAMQQueue<E extends
private long _flowResumeCapacity;
+ private ExclusivityPolicy _exclusivityPolicy;
+ private LifetimePolicy _lifetimePolicy;
+ private Object _exclusiveOwner; // could be connection, session or Principal
+
private final Set<NotificationCheck> _notificationChecks = EnumSet.noneOf(NotificationCheck.class);
@@ -157,7 +160,8 @@ abstract class SimpleAMQQueue<E extends
private final Set<AMQSessionModel> _blockedChannels = new ConcurrentSkipListSet<AMQSessionModel>();
private final AtomicBoolean _deleted = new AtomicBoolean(false);
- private final List<Action<AMQQueue>> _deleteTaskList = new CopyOnWriteArrayList<Action<AMQQueue>>();
+ private final List<Action<? super Q>> _deleteTaskList =
+ new CopyOnWriteArrayList<Action<? super Q>>();
private LogSubject _logSubject;
@@ -184,16 +188,98 @@ abstract class SimpleAMQQueue<E extends
private AMQQueue.NotificationListener _notificationListener;
private final long[] _lastNotificationTimes = new long[NotificationCheck.values().length];
-
- protected SimpleAMQQueue(UUID id,
- String name,
- boolean durable,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- VirtualHost virtualHost,
- QueueEntryListFactory<E,Q,L> entryListFactory, Map<String,Object> arguments)
+ protected SimpleAMQQueue(VirtualHost virtualHost,
+ final AMQSessionModel<?,?> creatingSession,
+ Map<String, Object> attributes,
+ QueueEntryListFactory<E, Q, L> entryListFactory)
{
+ UUID id = MapValueConverter.getUUIDAttribute(Queue.ID, attributes);
+ String name = MapValueConverter.getStringAttribute(Queue.NAME, attributes);
+ boolean durable = MapValueConverter.getBooleanAttribute(Queue.DURABLE,attributes,false);
+
+
+ _exclusivityPolicy = MapValueConverter.getEnumAttribute(ExclusivityPolicy.class,
+ Queue.EXCLUSIVE,
+ attributes,
+ ExclusivityPolicy.NONE);
+ _lifetimePolicy = MapValueConverter.getEnumAttribute(LifetimePolicy.class,
+ Queue.LIFETIME_POLICY,
+ attributes,
+ LifetimePolicy.PERMANENT);
+ if(creatingSession != null)
+ {
+
+ switch(_exclusivityPolicy)
+ {
+
+ case PRINCIPAL:
+ _exclusiveOwner = creatingSession.getConnectionModel().getAuthorizedPrincipal();
+ break;
+ case CONTAINER:
+ _exclusiveOwner = creatingSession.getConnectionModel().getRemoteContainerName();
+ break;
+ case CONNECTION:
+ _exclusiveOwner = creatingSession.getConnectionModel();
+ addExclusivityConstraint(creatingSession.getConnectionModel());
+ break;
+ case SESSION:
+ _exclusiveOwner = creatingSession;
+ addExclusivityConstraint(creatingSession);
+ break;
+ case NONE:
+ case LINK:
+ // nothing to do as if link no link associated until there is a consumer associated
+ break;
+ default:
+ throw new ServerScopedRuntimeException("Unknown exclusivity policy: "
+ + _exclusivityPolicy
+ + " this is a coding error inside Qpid");
+ }
+ }
+ else if(_exclusivityPolicy == ExclusivityPolicy.PRINCIPAL)
+ {
+ String owner = MapValueConverter.getStringAttribute(Queue.OWNER, attributes, null);
+ if(owner != null)
+ {
+ _exclusiveOwner = new AuthenticatedPrincipal(owner);
+ }
+ }
+ else if(_exclusivityPolicy == ExclusivityPolicy.CONTAINER)
+ {
+ String owner = MapValueConverter.getStringAttribute(Queue.OWNER, attributes, null);
+ if(owner != null)
+ {
+ _exclusiveOwner = owner;
+ }
+ }
+
+
+ if(_lifetimePolicy == LifetimePolicy.DELETE_ON_CONNECTION_CLOSE)
+ {
+ if(creatingSession != null)
+ {
+ addLifetimeConstraint(creatingSession.getConnectionModel());
+ }
+ else
+ {
+ throw new IllegalArgumentException("Queues created with a lifetime policy of "
+ + _lifetimePolicy
+ + " must be created from a connection.");
+ }
+ }
+ else if(_lifetimePolicy == LifetimePolicy.DELETE_ON_SESSION_END)
+ {
+ if(creatingSession != null)
+ {
+ addLifetimeConstraint(creatingSession);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Queues created with a lifetime policy of "
+ + _lifetimePolicy
+ + " must be created from a connection.");
+ }
+ }
if (name == null)
{
@@ -207,12 +293,18 @@ abstract class SimpleAMQQueue<E extends
_name = name;
_durable = durable;
- _owner = owner;
- _autoDelete = autoDelete;
- _exclusive = exclusive;
_virtualHost = virtualHost;
- _entries = entryListFactory.createQueueEntryList((Q)this);
- _arguments = Collections.synchronizedMap(arguments == null ? new LinkedHashMap<String, Object>() : new LinkedHashMap<String, Object>(arguments));
+ _entries = entryListFactory.createQueueEntryList((Q) this);
+ final LinkedHashMap<String, Object> arguments = new LinkedHashMap<String, Object>(attributes);
+
+ arguments.put(Queue.EXCLUSIVE, _exclusivityPolicy);
+ arguments.put(Queue.LIFETIME_POLICY, _lifetimePolicy);
+
+ _arguments = Collections.synchronizedMap(arguments);
+ _description = MapValueConverter.getStringAttribute(Queue.DESCRIPTION, attributes, null);
+
+ _noLocal = MapValueConverter.getBooleanAttribute(Queue.NO_LOCAL, attributes, false);
+
_id = id;
_asyncDelivery = ReferenceCountingExecutorService.getInstance().acquireExecutorService();
@@ -220,30 +312,113 @@ abstract class SimpleAMQQueue<E extends
_logSubject = new QueueLogSubject(this);
_logActor = new QueueActor(this, CurrentActor.get().getRootMessageLogger());
+
+ if (attributes.containsKey(Queue.ALERT_THRESHOLD_MESSAGE_AGE))
+ {
+ setMaximumMessageAge(MapValueConverter.getLongAttribute(Queue.ALERT_THRESHOLD_MESSAGE_AGE, attributes));
+ }
+ else
+ {
+ setMaximumMessageAge(virtualHost.getDefaultAlertThresholdMessageAge());
+ }
+ if (attributes.containsKey(Queue.ALERT_THRESHOLD_MESSAGE_SIZE))
+ {
+ setMaximumMessageSize(MapValueConverter.getLongAttribute(Queue.ALERT_THRESHOLD_MESSAGE_SIZE, attributes));
+ }
+ else
+ {
+ setMaximumMessageSize(virtualHost.getDefaultAlertThresholdMessageSize());
+ }
+ if (attributes.containsKey(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES))
+ {
+ setMaximumMessageCount(MapValueConverter.getLongAttribute(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES,
+ attributes));
+ }
+ else
+ {
+ setMaximumMessageCount(virtualHost.getDefaultAlertThresholdQueueDepthMessages());
+ }
+ if (attributes.containsKey(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_BYTES))
+ {
+ setMaximumQueueDepth(MapValueConverter.getLongAttribute(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_BYTES,
+ attributes));
+ }
+ else
+ {
+ setMaximumQueueDepth(virtualHost.getDefaultAlertThresholdQueueDepthBytes());
+ }
+ if (attributes.containsKey(Queue.ALERT_REPEAT_GAP))
+ {
+ setMinimumAlertRepeatGap(MapValueConverter.getLongAttribute(Queue.ALERT_REPEAT_GAP, attributes));
+ }
+ else
+ {
+ setMinimumAlertRepeatGap(virtualHost.getDefaultAlertRepeatGap());
+ }
+ if (attributes.containsKey(Queue.QUEUE_FLOW_CONTROL_SIZE_BYTES))
+ {
+ setCapacity(MapValueConverter.getLongAttribute(Queue.QUEUE_FLOW_CONTROL_SIZE_BYTES, attributes));
+ }
+ else
+ {
+ setCapacity(virtualHost.getDefaultQueueFlowControlSizeBytes());
+ }
+ if (attributes.containsKey(Queue.QUEUE_FLOW_RESUME_SIZE_BYTES))
+ {
+ setFlowResumeCapacity(MapValueConverter.getLongAttribute(Queue.QUEUE_FLOW_RESUME_SIZE_BYTES, attributes));
+ }
+ else
+ {
+ setFlowResumeCapacity(virtualHost.getDefaultQueueFlowResumeSizeBytes());
+ }
+ if (attributes.containsKey(Queue.MAXIMUM_DELIVERY_ATTEMPTS))
+ {
+ setMaximumDeliveryCount(MapValueConverter.getIntegerAttribute(Queue.MAXIMUM_DELIVERY_ATTEMPTS, attributes));
+ }
+ else
+ {
+ setMaximumDeliveryCount(virtualHost.getDefaultMaximumDeliveryAttempts());
+ }
+
+ final String ownerString;
+ switch(_exclusivityPolicy)
+ {
+ case PRINCIPAL:
+ ownerString = ((Principal) _exclusiveOwner).getName();
+ break;
+ case CONTAINER:
+ ownerString = (String) _exclusiveOwner;
+ break;
+ default:
+ ownerString = null;
+
+ }
+
// Log the creation of this Queue.
// The priorities display is toggled on if we set priorities > 0
CurrentActor.get().message(_logSubject,
- QueueMessages.CREATED(String.valueOf(_owner),
+ QueueMessages.CREATED(ownerString,
_entries.getPriorities(),
- _owner != null,
- autoDelete,
- durable, !durable,
+ ownerString != null ,
+ _lifetimePolicy != LifetimePolicy.PERMANENT,
+ durable,
+ !durable,
_entries.getPriorities() > 0));
- if(arguments != null && arguments.containsKey(Queue.MESSAGE_GROUP_KEY))
+ if(attributes != null && attributes.containsKey(Queue.MESSAGE_GROUP_KEY))
{
- if(arguments.get(Queue.MESSAGE_GROUP_SHARED_GROUPS) != null
- && (Boolean)(arguments.get(Queue.MESSAGE_GROUP_SHARED_GROUPS)))
+ if(attributes.get(Queue.MESSAGE_GROUP_SHARED_GROUPS) != null
+ && (Boolean)(attributes.get(Queue.MESSAGE_GROUP_SHARED_GROUPS)))
{
- Object defaultGroup = arguments.get(Queue.MESSAGE_GROUP_DEFAULT_GROUP);
+ Object defaultGroup = attributes.get(Queue.MESSAGE_GROUP_DEFAULT_GROUP);
_messageGroupManager =
- new DefinedGroupMessageGroupManager<E,Q,L>(String.valueOf(arguments.get(Queue.MESSAGE_GROUP_KEY)),
+ new DefinedGroupMessageGroupManager<E,Q,L>(String.valueOf(attributes.get(Queue.MESSAGE_GROUP_KEY)),
defaultGroup == null ? DEFAULT_SHARED_MESSAGE_GROUP : defaultGroup.toString(),
this);
}
else
{
- _messageGroupManager = new AssignedConsumerMessageGroupManager<E,Q,L>(String.valueOf(arguments.get(
+ _messageGroupManager = new AssignedConsumerMessageGroupManager<E,Q,L>(String.valueOf(attributes.get(
Queue.MESSAGE_GROUP_KEY)), DEFAULT_MAX_GROUPS);
}
}
@@ -256,6 +431,38 @@ abstract class SimpleAMQQueue<E extends
}
+ private void addLifetimeConstraint(final Deletable<? extends Deletable> lifetimeObject)
+ {
+ final Action<Deletable> deleteQueueTask = new Action<Deletable>()
+ {
+ @Override
+ public void performAction(final Deletable object)
+ {
+ try
+ {
+ getVirtualHost().removeQueue(SimpleAMQQueue.this);
+ }
+ catch (QpidSecurityException e)
+ {
+ throw new ConnectionScopedRuntimeException("Unable to delete a queue even though the queue's " +
+ "lifetime was tied to an object being deleted");
+ }
+ }
+ };
+
+ lifetimeObject.addDeleteTask(deleteQueueTask);
+ addDeleteTask(new DeleteDeleteTask(lifetimeObject, deleteQueueTask));
+ }
+
+ private void addExclusivityConstraint(final Deletable<? extends Deletable> lifetimeObject)
+ {
+ final ClearOwnerAction clearOwnerAction = new ClearOwnerAction(lifetimeObject);
+ final DeleteDeleteTask deleteDeleteTask = new DeleteDeleteTask(lifetimeObject, clearOwnerAction);
+ clearOwnerAction.setDeleteTask(deleteDeleteTask);
+ lifetimeObject.addDeleteTask(clearOwnerAction);
+ addDeleteTask(deleteDeleteTask);
+ }
+
public void resetNotifications()
{
// This ensure that the notification checks for the configured alerts are created.
@@ -303,12 +510,7 @@ abstract class SimpleAMQQueue<E extends
public boolean isExclusive()
{
- return _exclusive;
- }
-
- public void setExclusive(boolean exclusive)
- {
- _exclusive = exclusive;
+ return _exclusivityPolicy != ExclusivityPolicy.NONE;
}
public Exchange getAlternateExchange()
@@ -342,27 +544,27 @@ abstract class SimpleAMQQueue<E extends
return _arguments.get(attrName);
}
- public boolean isAutoDelete()
+ @Override
+ public LifetimePolicy getLifetimePolicy()
{
- return _autoDelete;
+ return _lifetimePolicy;
}
public String getOwner()
{
- return _owner;
- }
-
- public AuthorizationHolder getAuthorizationHolder()
- {
- return _authorizationHolder;
- }
-
- public void setAuthorizationHolder(final AuthorizationHolder authorizationHolder)
- {
- _authorizationHolder = authorizationHolder;
+ if(_exclusiveOwner != null)
+ {
+ switch(_exclusivityPolicy)
+ {
+ case CONTAINER:
+ return (String) _exclusiveOwner;
+ case PRINCIPAL:
+ return ((Principal)_exclusiveOwner).getName();
+ }
+ }
+ return null;
}
-
public VirtualHost getVirtualHost()
{
return _virtualHost;
@@ -381,7 +583,9 @@ abstract class SimpleAMQQueue<E extends
final FilterManager filters,
final Class<? extends ServerMessage> messageClass,
final String consumerName,
- EnumSet<Consumer.Option> optionSet) throws ExistingExclusiveConsumer, ExistingConsumerPreventsExclusive, QpidSecurityException
+ EnumSet<Consumer.Option> optionSet)
+ throws ExistingExclusiveConsumer, ExistingConsumerPreventsExclusive, QpidSecurityException,
+ ConsumerAccessRefused
{
// Access control
@@ -396,15 +600,77 @@ abstract class SimpleAMQQueue<E extends
throw new ExistingExclusiveConsumer();
}
+ switch(_exclusivityPolicy)
+ {
+ case CONNECTION:
+ if(_exclusiveOwner == null)
+ {
+ _exclusiveOwner = target.getSessionModel().getConnectionModel();
+ addExclusivityConstraint(target.getSessionModel().getConnectionModel());
+ }
+ else
+ {
+ if(_exclusiveOwner != target.getSessionModel().getConnectionModel())
+ {
+ throw new ConsumerAccessRefused();
+ }
+ }
+ break;
+ case SESSION:
+ if(_exclusiveOwner == null)
+ {
+ _exclusiveOwner = target.getSessionModel();
+ addExclusivityConstraint(target.getSessionModel());
+ }
+ else
+ {
+ if(_exclusiveOwner != target.getSessionModel())
+ {
+ throw new ConsumerAccessRefused();
+ }
+ }
+ break;
+ case LINK:
+ if(getConsumerCount() != 0)
+ {
+ throw new ConsumerAccessRefused();
+ }
+ break;
+ case PRINCIPAL:
+ if(_exclusiveOwner == null)
+ {
+ _exclusiveOwner = target.getSessionModel().getConnectionModel().getAuthorizedPrincipal();
+ }
+ else
+ {
+ if(!_exclusiveOwner.equals(target.getSessionModel().getConnectionModel().getAuthorizedPrincipal()))
+ {
+ throw new ConsumerAccessRefused();
+ }
+ }
+ break;
+ case CONTAINER:
+ if(_exclusiveOwner == null)
+ {
+ _exclusiveOwner = target.getSessionModel().getConnectionModel().getRemoteContainerName();
+ }
+ else
+ {
+ if(!_exclusiveOwner.equals(target.getSessionModel().getConnectionModel().getRemoteContainerName()))
+ {
+ throw new ConsumerAccessRefused();
+ }
+ }
+ break;
+ case NONE:
+ break;
+ default:
+ throw new ServerScopedRuntimeException("Unknown exclusivity policy " + _exclusivityPolicy);
+ }
boolean exclusive = optionSet.contains(Consumer.Option.EXCLUSIVE);
boolean isTransient = optionSet.contains(Consumer.Option.TRANSIENT);
- if (exclusive && !isTransient && getConsumerCount() != 0)
- {
- throw new ExistingConsumerPreventsExclusive();
- }
-
QueueConsumer<T,E,Q,L> consumer = new QueueConsumer<T,E,Q,L>(filters, messageClass,
optionSet.contains(Consumer.Option.ACQUIRES),
optionSet.contains(Consumer.Option.SEES_REQUEUES),
@@ -473,11 +739,12 @@ abstract class SimpleAMQQueue<E extends
consumer.close();
// No longer can the queue have an exclusive consumer
setExclusiveSubscriber(null);
+
consumer.setQueueContext(null);
- if(!isDeleted() && isExclusive() && getConsumerCount() == 0)
+ if(_exclusivityPolicy == ExclusivityPolicy.LINK)
{
- setAuthorizationHolder(null);
+ _exclusiveOwner = null;
}
if(_messageGroupManager != null)
@@ -495,8 +762,12 @@ abstract class SimpleAMQQueue<E extends
// auto-delete queues must be deleted if there are no remaining subscribers
- if (_autoDelete && getDeleteOnNoConsumers() && !consumer.isTransient() && getConsumerCount() == 0 )
+ if(!consumer.isTransient()
+ && ( _lifetimePolicy == LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS
+ || _lifetimePolicy == LifetimePolicy.DELETE_ON_NO_LINKS )
+ && getConsumerCount() == 0)
{
+
if (_logger.isInfoEnabled())
{
_logger.info("Auto-deleting queue:" + this);
@@ -1266,12 +1537,14 @@ abstract class SimpleAMQQueue<E extends
});
}
- public void addQueueDeleteTask(final Action<AMQQueue> task)
+ @Override
+ public void addDeleteTask(final Action<? super Q> task)
{
_deleteTaskList.add(task);
}
- public void removeQueueDeleteTask(final Action<AMQQueue> task)
+ @Override
+ public void removeDeleteTask(final Action<? super Q> task)
{
_deleteTaskList.remove(task);
}
@@ -1343,9 +1616,9 @@ abstract class SimpleAMQQueue<E extends
}
- for (Action<AMQQueue> task : _deleteTaskList)
+ for (Action<? super Q> task : _deleteTaskList)
{
- task.performAction(this);
+ task.performAction((Q)this);
}
_deleteTaskList.clear();
@@ -1940,6 +2213,26 @@ abstract class SimpleAMQQueue<E extends
return _notificationChecks;
}
+ private static class DeleteDeleteTask implements Action<Deletable>
+ {
+
+ private final Deletable<? extends Deletable> _lifetimeObject;
+ private final Action<? super Deletable> _deleteQueueOwnerTask;
+
+ public DeleteDeleteTask(final Deletable<? extends Deletable> lifetimeObject,
+ final Action<? super Deletable> deleteQueueOwnerTask)
+ {
+ _lifetimeObject = lifetimeObject;
+ _deleteQueueOwnerTask = deleteQueueOwnerTask;
+ }
+
+ @Override
+ public void performAction(final Deletable object)
+ {
+ _lifetimeObject.removeDeleteTask(_deleteQueueOwnerTask);
+ }
+ }
+
private final class QueueEntryListener implements StateChangeListener<E, QueueEntry.State>
{
@@ -1990,38 +2283,6 @@ abstract class SimpleAMQQueue<E extends
return ids;
}
- public AMQSessionModel getExclusiveOwningSession()
- {
- return _exclusiveOwner;
- }
-
- public void setExclusiveOwningSession(AMQSessionModel exclusiveOwner)
- {
- _exclusive = true;
- _exclusiveOwner = exclusiveOwner;
- }
-
-
- public void configure(QueueConfiguration config)
- {
- if (config != null)
- {
- setMaximumMessageAge(config.getMaximumMessageAge());
- setMaximumQueueDepth(config.getMaximumQueueDepth());
- setMaximumMessageSize(config.getMaximumMessageSize());
- setMaximumMessageCount(config.getMaximumMessageCount());
- setMinimumAlertRepeatGap(config.getMinimumAlertRepeatGap());
- setMaximumDeliveryCount(config.getMaxDeliveryCount());
- _capacity = config.getCapacity();
- _flowResumeCapacity = config.getFlowResumeCapacity();
- }
- }
-
- public long getMessageDequeueCount()
- {
- return _dequeueCount.get();
- }
-
public long getTotalEnqueueSize()
{
return _enqueueSize.get();
@@ -2130,20 +2391,13 @@ abstract class SimpleAMQQueue<E extends
@Override
public void setDescription(String description)
{
- if (description == null)
- {
- _arguments.remove(Queue.DESCRIPTION);
- }
- else
- {
- _arguments.put(Queue.DESCRIPTION, description);
- }
+ _description = description;
}
@Override
public String getDescription()
{
- return (String) _arguments.get(Queue.DESCRIPTION);
+ return _description;
}
public final <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
@@ -2176,4 +2430,228 @@ abstract class SimpleAMQQueue<E extends
}
+ @Override
+ public boolean verifySessionAccess(final AMQSessionModel<?, ?> session)
+ {
+ boolean allowed;
+ switch(_exclusivityPolicy)
+ {
+ case NONE:
+ allowed = true;
+ break;
+ case SESSION:
+ allowed = _exclusiveOwner == null || _exclusiveOwner == session;
+ break;
+ case CONNECTION:
+ allowed = _exclusiveOwner == null || _exclusiveOwner == session.getConnectionModel();
+ break;
+ case PRINCIPAL:
+ allowed = _exclusiveOwner == null || _exclusiveOwner.equals(session.getConnectionModel().getAuthorizedPrincipal());
+ break;
+ case CONTAINER:
+ allowed = _exclusiveOwner == null || _exclusiveOwner.equals(session.getConnectionModel().getRemoteContainerName());
+ break;
+ case LINK:
+ allowed = _exclusiveSubscriber == null || _exclusiveSubscriber.getSessionModel() == session;
+ break;
+ default:
+ throw new ServerScopedRuntimeException("Unknown exclusivity policy " + _exclusivityPolicy);
+ }
+ return allowed;
+ }
+
+ @Override
+ public synchronized void setExclusivityPolicy(final ExclusivityPolicy desiredPolicy)
+ throws ExistingConsumerPreventsExclusive
+ {
+ if(desiredPolicy != _exclusivityPolicy && !(desiredPolicy == null && _exclusivityPolicy == ExclusivityPolicy.NONE))
+ {
+ switch(desiredPolicy)
+ {
+ case NONE:
+ _exclusiveOwner = null;
+ break;
+ case PRINCIPAL:
+ switchToPrincipalExclusivity();
+ break;
+ case CONTAINER:
+ switchToContainerExclusivity();
+ break;
+ case CONNECTION:
+ switchToConnectionExclusivity();
+ break;
+ case SESSION:
+ switchToSessionExclusivity();
+ break;
+ case LINK:
+ switchToLinkExclusivity();
+ break;
+ }
+ _exclusivityPolicy = desiredPolicy;
+ }
+ }
+
+ private void switchToLinkExclusivity() throws ExistingConsumerPreventsExclusive
+ {
+ switch (getConsumerCount())
+ {
+ case 1:
+ _exclusiveSubscriber = getConsumerList().getHead().getConsumer();
+ // deliberate fall through
+ case 0:
+ _exclusiveOwner = null;
+ break;
+ default:
+ throw new ExistingConsumerPreventsExclusive();
+ }
+
+ }
+
+ private void switchToSessionExclusivity() throws ExistingConsumerPreventsExclusive
+ {
+
+ switch(_exclusivityPolicy)
+ {
+ case NONE:
+ case PRINCIPAL:
+ case CONTAINER:
+ case CONNECTION:
+ AMQSessionModel session = null;
+ for(Consumer c : getConsumers())
+ {
+ if(session == null)
+ {
+ session = c.getSessionModel();
+ }
+ else if(!session.equals(c.getSessionModel()))
+ {
+ throw new ExistingConsumerPreventsExclusive();
+ }
+ }
+ _exclusiveOwner = session;
+ break;
+ case LINK:
+ _exclusiveOwner = _exclusiveSubscriber == null ? null : _exclusiveSubscriber.getSessionModel().getConnectionModel();
+ }
+ }
+
+ private void switchToConnectionExclusivity() throws ExistingConsumerPreventsExclusive
+ {
+ switch(_exclusivityPolicy)
+ {
+ case NONE:
+ case CONTAINER:
+ case PRINCIPAL:
+ AMQConnectionModel con = null;
+ for(Consumer c : getConsumers())
+ {
+ if(con == null)
+ {
+ con = c.getSessionModel().getConnectionModel();
+ }
+ else if(!con.equals(c.getSessionModel().getConnectionModel()))
+ {
+ throw new ExistingConsumerPreventsExclusive();
+ }
+ }
+ _exclusiveOwner = con;
+ break;
+ case SESSION:
+ _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQSessionModel)_exclusiveOwner).getConnectionModel();
+ break;
+ case LINK:
+ _exclusiveOwner = _exclusiveSubscriber == null ? null : _exclusiveSubscriber.getSessionModel().getConnectionModel();
+ }
+ }
+
+ private void switchToContainerExclusivity() throws ExistingConsumerPreventsExclusive
+ {
+ switch(_exclusivityPolicy)
+ {
+ case NONE:
+ case PRINCIPAL:
+ String containerID = null;
+ for(Consumer c : getConsumers())
+ {
+ if(containerID == null)
+ {
+ containerID = c.getSessionModel().getConnectionModel().getRemoteContainerName();
+ }
+ else if(!containerID.equals(c.getSessionModel().getConnectionModel().getRemoteContainerName()))
+ {
+ throw new ExistingConsumerPreventsExclusive();
+ }
+ }
+ _exclusiveOwner = containerID;
+ break;
+ case CONNECTION:
+ _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQConnectionModel)_exclusiveOwner).getRemoteContainerName();
+ break;
+ case SESSION:
+ _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQSessionModel)_exclusiveOwner).getConnectionModel().getRemoteContainerName();
+ break;
+ case LINK:
+ _exclusiveOwner = _exclusiveSubscriber == null ? null : _exclusiveSubscriber.getSessionModel().getConnectionModel().getRemoteContainerName();
+ }
+ }
+
+ private void switchToPrincipalExclusivity() throws ExistingConsumerPreventsExclusive
+ {
+ switch(_exclusivityPolicy)
+ {
+ case NONE:
+ case CONTAINER:
+ Principal principal = null;
+ for(Consumer c : getConsumers())
+ {
+ if(principal == null)
+ {
+ principal = c.getSessionModel().getConnectionModel().getAuthorizedPrincipal();
+ }
+ else if(!principal.equals(c.getSessionModel().getConnectionModel().getAuthorizedPrincipal()))
+ {
+ throw new ExistingConsumerPreventsExclusive();
+ }
+ }
+ _exclusiveOwner = principal;
+ break;
+ case CONNECTION:
+ _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQConnectionModel)_exclusiveOwner).getAuthorizedPrincipal();
+ break;
+ case SESSION:
+ _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQSessionModel)_exclusiveOwner).getConnectionModel().getAuthorizedPrincipal();
+ break;
+ case LINK:
+ _exclusiveOwner = _exclusiveSubscriber == null ? null : _exclusiveSubscriber.getSessionModel().getConnectionModel().getAuthorizedPrincipal();
+ }
+ }
+
+ private class ClearOwnerAction implements Action<Deletable>
+ {
+ private final Deletable<? extends Deletable> _lifetimeObject;
+ private DeleteDeleteTask _deleteTask;
+
+ public ClearOwnerAction(final Deletable<? extends Deletable> lifetimeObject)
+ {
+ _lifetimeObject = lifetimeObject;
+ }
+
+ @Override
+ public void performAction(final Deletable object)
+ {
+ if(SimpleAMQQueue.this._exclusiveOwner == _lifetimeObject)
+ {
+ SimpleAMQQueue.this._exclusiveOwner = null;
+ }
+ if(_deleteTask != null)
+ {
+ removeDeleteTask(_deleteTask);
+ }
+ }
+
+ public void setDeleteTask(final DeleteDeleteTask deleteTask)
+ {
+ _deleteTask = deleteTask;
+ }
+ }
}
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java Mon Feb 17 20:19:36 2014
@@ -21,11 +21,13 @@ package org.apache.qpid.server.queue;
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.util.Action;
+import org.apache.qpid.server.util.MapValueConverter;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Map;
-import java.util.UUID;
public class SortedQueue extends OutOfOrderQueue<SortedQueueEntry, SortedQueue, SortedQueueEntryList>
{
@@ -35,28 +37,26 @@ public class SortedQueue extends OutOfOr
private final Object _sortedQueueLock = new Object();
private final String _sortedPropertyName;
- protected SortedQueue(UUID id, final String name,
- final boolean durable, final String owner, final boolean autoDelete,
- final boolean exclusive, final VirtualHost virtualHost, Map<String, Object> arguments, String sortedPropertyName)
+ protected SortedQueue(VirtualHost virtualHost,
+ final AMQSessionModel creatingSession,
+ Map<String, Object> attributes,
+ QueueEntryListFactory<SortedQueueEntry, SortedQueue, SortedQueueEntryList> factory)
{
- this(id, name, durable, owner, autoDelete, exclusive,
- virtualHost, arguments, sortedPropertyName, new SortedQueueEntryListFactory(sortedPropertyName));
+ super(virtualHost, creatingSession, attributes, factory);
+ _sortedPropertyName = MapValueConverter.getStringAttribute(Queue.SORT_KEY,attributes);
}
- protected SortedQueue(UUID id, final String name,
- final boolean durable, final String owner, final boolean autoDelete,
- final boolean exclusive, final VirtualHost virtualHost,
- Map<String, Object> arguments,
- String sortedPropertyName,
- QueueEntryListFactory<SortedQueueEntry,SortedQueue,SortedQueueEntryList> factory)
+ protected SortedQueue(VirtualHost virtualHost,
+ final AMQSessionModel creatingSession, Map<String, Object> attributes)
{
- super(id, name, durable, owner, autoDelete, exclusive,
- virtualHost, factory, arguments);
- this._sortedPropertyName = sortedPropertyName;
+ this(virtualHost,
+ creatingSession, attributes,
+ new SortedQueueEntryListFactory(MapValueConverter.getStringAttribute(Queue.SORT_KEY, attributes)));
}
+
public String getSortedPropertyName()
{
return _sortedPropertyName;
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/StandardQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/StandardQueue.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/StandardQueue.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/StandardQueue.java Mon Feb 17 20:19:36 2014
@@ -20,22 +20,16 @@
*/
package org.apache.qpid.server.queue;
+import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Map;
-import java.util.UUID;
public class StandardQueue extends SimpleAMQQueue<StandardQueueEntry,StandardQueue,StandardQueueEntryList>
{
- public StandardQueue(final UUID id,
- final String name,
- final boolean durable,
- final String owner,
- final boolean autoDelete,
- final boolean exclusive,
- final VirtualHost virtualHost,
- final Map<String, Object> arguments)
+ public StandardQueue(final VirtualHost virtualHost,
+ final AMQSessionModel creatingSession, final Map<String, Object> arguments)
{
- super(id, name, durable, owner, autoDelete, exclusive, virtualHost, new StandardQueueEntryList.Factory(), arguments);
+ super(virtualHost, creatingSession, arguments, new StandardQueueEntryList.Factory());
}
}
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java Mon Feb 17 20:19:36 2014
@@ -27,6 +27,7 @@ import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.queue.AMQQueue;
/**
@@ -139,8 +140,8 @@ public class ObjectProperties
{
setName(queue.getName());
- put(Property.AUTO_DELETE, queue.isAutoDelete());
- put(Property.TEMPORARY, queue.isAutoDelete());
+ put(Property.AUTO_DELETE, queue.getLifetimePolicy() != LifetimePolicy.PERMANENT);
+ put(Property.TEMPORARY, queue.getLifetimePolicy() != LifetimePolicy.PERMANENT);
put(Property.DURABLE, queue.isDurable());
put(Property.EXCLUSIVE, queue.isExclusive());
if (queue.getAlternateExchange() != null)
@@ -151,10 +152,7 @@ public class ObjectProperties
{
put(Property.OWNER, queue.getOwner());
}
- else if (queue.getAuthorizationHolder() != null)
- {
- put(Property.OWNER, queue.getAuthorizationHolder().getAuthorizedPrincipal().getName());
- }
+
}
public ObjectProperties(Exchange exch, AMQQueue queue, String routingKey)
Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java?rev=1569102&r1=1569101&r2=1569102&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java Mon Feb 17 20:19:36 2014
@@ -42,17 +42,11 @@ public class DurableConfigurationStoreHe
private static final String BINDING = Binding.class.getSimpleName();
private static final String EXCHANGE = Exchange.class.getSimpleName();
private static final String QUEUE = Queue.class.getSimpleName();
- private static final Set<String> QUEUE_ARGUMENTS_EXCLUDES = new HashSet<String>(Arrays.asList(Queue.NAME,
- Queue.OWNER,
- Queue.EXCLUSIVE,
- Queue.ALTERNATE_EXCHANGE));
+ private static final Set<String> QUEUE_ARGUMENTS_EXCLUDES = new HashSet<String>(Arrays.asList(Queue.ALTERNATE_EXCHANGE));
public static void updateQueue(DurableConfigurationStore store, AMQQueue<?,?,?> queue)
{
Map<String, Object> attributesMap = new LinkedHashMap<String, Object>();
- attributesMap.put(Queue.NAME, queue.getName());
- attributesMap.put(Queue.OWNER, queue.getOwner());
- attributesMap.put(Queue.EXCLUSIVE, queue.isExclusive());
if (queue.getAlternateExchange() != null)
{
@@ -75,9 +69,6 @@ public class DurableConfigurationStoreHe
public static void createQueue(DurableConfigurationStore store, AMQQueue<?,?,?> queue)
{
Map<String, Object> attributesMap = new HashMap<String, Object>();
- attributesMap.put(Queue.NAME, queue.getName());
- attributesMap.put(Queue.OWNER, queue.getOwner());
- attributesMap.put(Queue.EXCLUSIVE, queue.isExclusive());
if (queue.getAlternateExchange() != null)
{
attributesMap.put(Queue.ALTERNATE_EXCHANGE, queue.getAlternateExchange().getId());
@@ -103,7 +94,7 @@ public class DurableConfigurationStoreHe
Map<String, Object> attributesMap = new HashMap<String, Object>();
attributesMap.put(Exchange.NAME, exchange.getName());
attributesMap.put(Exchange.TYPE, exchange.getTypeName());
- attributesMap.put(Exchange.LIFETIME_POLICY, exchange.isAutoDelete() ? LifetimePolicy.AUTO_DELETE.name()
+ attributesMap.put(Exchange.LIFETIME_POLICY, exchange.isAutoDelete() ? LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS.name()
: LifetimePolicy.PERMANENT.name());
store.create(exchange.getId(), EXCHANGE, attributesMap);
Added: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/util/Deletable.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/util/Deletable.java?rev=1569102&view=auto
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/util/Deletable.java (added)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/util/Deletable.java Mon Feb 17 20:19:36 2014
@@ -0,0 +1,27 @@
+/*
+ *
+ * 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.qpid.server.util;
+
+public interface Deletable<T extends Deletable>
+{
+ void addDeleteTask(Action<? super T> task);
+ void removeDeleteTask(Action<? super T> task);
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org