Update Queue implementation to better define lifetime and exclusivity policies
git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/java-broker-amqp-1-0-management@1569102 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
index 270abef..1556876 100644
--- a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
+++ b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
@@ -563,7 +563,7 @@
{
_receivingSessions[channel] = null;
- endpoint.end(end);
+ endpoint.receiveEnd(end);
}
else
{
diff --git a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java
index c37c52c..c9212b1 100644
--- a/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java
+++ b/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SessionEndpoint.java
@@ -153,19 +153,47 @@
public void end()
{
- end(null);
+ end(new End());
}
- public void end(final End end)
+ public void end(End end)
{
synchronized(getLock())
{
switch(_state)
{
case BEGIN_SENT:
- _connection.sendEnd(getSendingChannel(), new End(), false);
+ _connection.sendEnd(getSendingChannel(), end, false);
_state = SessionState.END_PIPE;
break;
+ case ACTIVE:
+ detachLinks();
+ short sendChannel = getSendingChannel();
+ _connection.sendEnd(sendChannel, end, true);
+ _state = SessionState.END_SENT;
+ break;
+ default:
+ sendChannel = getSendingChannel();
+ End reply = new End();
+ Error error = new Error();
+ error.setCondition(AmqpError.ILLEGAL_STATE);
+ error.setDescription("END called on Session which has not been opened");
+ reply.setError(error);
+ _connection.sendEnd(sendChannel, reply, true);
+ break;
+
+
+ }
+ getLock().notifyAll();
+ }
+ }
+
+ public void receiveEnd(final End end)
+ {
+ synchronized(getLock())
+ {
+ switch(_state)
+ {
case END_SENT:
_state = SessionState.ENDED;
break;
@@ -174,7 +202,7 @@
_sessionEventListener.remoteEnd(end);
short sendChannel = getSendingChannel();
_connection.sendEnd(sendChannel, new End(), true);
- _state = end == null ? SessionState.END_SENT : SessionState.ENDED;
+ _state = SessionState.ENDED;
break;
default:
sendChannel = getSendingChannel();
diff --git a/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java b/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java
index 9b1ab70..46f2afd 100644
--- a/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java
+++ b/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java
@@ -568,7 +568,7 @@
Map<String, Object> attributesMap = new HashMap<String, Object>();
attributesMap.put(Exchange.NAME, exchangeName);
attributesMap.put(Exchange.TYPE, exchangeType);
- attributesMap.put(Exchange.LIFETIME_POLICY, autoDelete ? LifetimePolicy.AUTO_DELETE.name()
+ attributesMap.put(Exchange.LIFETIME_POLICY, autoDelete ? "AUTO_DELETE"
: LifetimePolicy.PERMANENT.name());
String json = _serializer.serialize(attributesMap);
UpgradeConfiguredObjectRecord configuredObject = new UpgradeConfiguredObjectRecord(Exchange.class.getName(), json);
diff --git a/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java b/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java
index 44f0861..0249400 100644
--- a/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java
+++ b/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6Test.java
@@ -44,6 +44,7 @@
import org.apache.log4j.Logger;
import org.apache.qpid.server.model.Binding;
import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.ExclusivityPolicy;
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.UUIDGenerator;
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/configuration/QueueConfiguration.java b/java/broker-core/src/main/java/org/apache/qpid/server/configuration/QueueConfiguration.java
index 25466d9..4e9d94c 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/configuration/QueueConfiguration.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/configuration/QueueConfiguration.java
@@ -20,7 +20,9 @@
*/
package org.apache.qpid.server.configuration;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.Iterator;
import java.util.Map;
import org.apache.commons.configuration.CompositeConfiguration;
import org.apache.commons.configuration.Configuration;
@@ -43,7 +45,6 @@
CompositeConfiguration mungedConf = new CompositeConfiguration();
mungedConf.addConfiguration(_vHostConfig.getConfig().subset("queues.queue." + escapeTagName(name)));
- mungedConf.addConfiguration(_vHostConfig.getConfig().subset("queues"));
setConfiguration("virtualhosts.virtualhost.queues.queue", mungedConf);
}
@@ -85,19 +86,33 @@
return _vHostConfig;
}
+ private boolean getDefaultedBoolean(String attribute)
+ {
+ final Configuration config = _vHostConfig.getConfig();
+ if(config.containsKey("queues."+attribute))
+ {
+ final boolean defaultValue = config.getBoolean("queues." + attribute);
+ return getBooleanValue(attribute, defaultValue);
+ }
+ else
+ {
+ return getBooleanValue(attribute);
+ }
+ }
+
public boolean getDurable()
{
- return getBooleanValue("durable");
+ return getDefaultedBoolean("boolean");
}
public boolean getExclusive()
{
- return getBooleanValue("exclusive");
+ return getDefaultedBoolean("exclusive");
}
public boolean getAutoDelete()
{
- return getBooleanValue("autodelete");
+ return getDefaultedBoolean("autodelete");
}
public String getOwner()
@@ -107,17 +122,41 @@
public boolean getPriority()
{
- return getBooleanValue("priority");
+ return getDefaultedBoolean("priority");
}
public int getPriorities()
{
- return getIntValue("priorities", -1);
+ final Configuration config = _vHostConfig.getConfig();
+
+ int defaultValue;
+ if(config.containsKey("queues.priorities"))
+ {
+ defaultValue = config.getInt("queues.priorities");
+ }
+ else
+ {
+ defaultValue = -1;
+ }
+ return getIntValue("priorities", defaultValue);
}
public String getExchange()
{
- return getStringValue("exchange", ExchangeDefaults.DEFAULT_EXCHANGE_NAME);
+ final Configuration config = _vHostConfig.getConfig();
+
+ String defaultValue;
+
+ if(config.containsKey("queues.exchange"))
+ {
+ defaultValue = config.getString("queues.exchange");
+ }
+ else
+ {
+ defaultValue = "";
+ }
+
+ return getStringValue("exchange", defaultValue);
}
public List getRoutingKeys()
@@ -137,37 +176,37 @@
public int getMaximumMessageAge()
{
- return getIntValue("maximumMessageAge", _vHostConfig.getMaximumMessageAge());
+ return getIntValue("maximumMessageAge");
}
public long getMaximumQueueDepth()
{
- return getLongValue("maximumQueueDepth", _vHostConfig.getMaximumQueueDepth());
+ return getLongValue("maximumQueueDepth");
}
public long getMaximumMessageSize()
{
- return getLongValue("maximumMessageSize", _vHostConfig.getMaximumMessageSize());
+ return getLongValue("maximumMessageSize");
}
public long getMaximumMessageCount()
{
- return getLongValue("maximumMessageCount", _vHostConfig.getMaximumMessageCount());
+ return getLongValue("maximumMessageCount");
}
public long getMinimumAlertRepeatGap()
{
- return getLongValue("minimumAlertRepeatGap", _vHostConfig.getMinimumAlertRepeatGap());
+ return getLongValue("minimumAlertRepeatGap");
}
public long getCapacity()
{
- return getLongValue("capacity", _vHostConfig.getCapacity());
+ return getLongValue("capacity");
}
public long getFlowResumeCapacity()
{
- return getLongValue("flowResumeCapacity", _vHostConfig.getFlowResumeCapacity());
+ return getLongValue("flowResumeCapacity");
}
public boolean isLVQ()
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/filter/FilterSupport.java b/java/broker-core/src/main/java/org/apache/qpid/server/filter/FilterSupport.java
index 9870551..ea0faf1 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/filter/FilterSupport.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/filter/FilterSupport.java
@@ -22,6 +22,7 @@
package org.apache.qpid.server.filter;
import java.lang.ref.WeakReference;
+import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.WeakHashMap;
@@ -29,6 +30,7 @@
import org.apache.qpid.filter.SelectorParsingException;
import org.apache.qpid.filter.selector.ParseException;
import org.apache.qpid.filter.selector.TokenMgrError;
+import org.apache.qpid.server.consumer.Consumer;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.queue.AMQQueue;
@@ -120,19 +122,25 @@
public static final class NoLocalFilter implements MessageFilter
{
- private final MessageSource _queue;
+ private final MessageSource<?,?> _queue;
- public NoLocalFilter(MessageSource queue)
+ private NoLocalFilter(MessageSource queue)
{
_queue = queue;
}
public boolean matches(Filterable message)
{
- final AMQSessionModel exclusiveOwningSession = _queue.getExclusiveOwningSession();
- return exclusiveOwningSession == null ||
- exclusiveOwningSession.getConnectionReference() != message.getConnectionReference();
+ final Collection<? extends Consumer> consumers = _queue.getConsumers();
+ for(Consumer c : consumers)
+ {
+ if(c.getSessionModel().getConnectionReference() == message.getConnectionReference())
+ {
+ return false;
+ }
+ }
+ return !consumers.isEmpty();
}
@Override
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java b/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
index 5b0e34b..dcbb693 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
@@ -46,7 +46,7 @@
AMQConnectionModel connection = session.getConnectionModel();
setLogStringWithFormat(CHANNEL_FORMAT,
connection == null ? -1L : connection.getConnectionId(),
- (connection == null || connection.getPrincipalAsString() == null) ? "?" : connection.getPrincipalAsString(),
+ (connection == null || connection.getAuthorizedPrincipal() == null) ? "?" : connection.getAuthorizedPrincipal().getName(),
(connection == null || connection.getRemoteAddressString() == null) ? "?" : connection.getRemoteAddressString(),
(connection == null || connection.getVirtualHostName() == null) ? "?" : connection.getVirtualHostName(),
session.getChannelId());
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java b/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java
index 87c2377..b1b8bc5 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java
@@ -55,7 +55,7 @@
{
if (!_upToDate)
{
- if (_session.getPrincipalAsString() != null)
+ if (_session.getAuthorizedPrincipal() != null)
{
if (_session.getVirtualHostName() != null)
{
@@ -71,7 +71,7 @@
*/
setLogString("[" + MessageFormat.format(CONNECTION_FORMAT,
_session.getConnectionId(),
- _session.getPrincipalAsString(),
+ _session.getAuthorizedPrincipal().getName(),
_session.getRemoteAddressString(),
_session.getVirtualHostName())
+ "] ");
@@ -82,7 +82,7 @@
{
setLogString("[" + MessageFormat.format(USER_FORMAT,
_session.getConnectionId(),
- _session.getPrincipalAsString(),
+ _session.getAuthorizedPrincipal().getName(),
_session.getRemoteAddressString())
+ "] ");
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java b/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java
index b947bc0..33c8e26 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java
@@ -36,7 +36,8 @@
<T extends ConsumerTarget> C addConsumer(T target, FilterManager filters,
Class<? extends ServerMessage> messageClass,
String consumerName, EnumSet<Consumer.Option> options)
- throws ExistingExclusiveConsumer, ExistingConsumerPreventsExclusive, QpidSecurityException;
+ throws ExistingExclusiveConsumer, ExistingConsumerPreventsExclusive, QpidSecurityException,
+ ConsumerAccessRefused;
Collection<C> getConsumers();
@@ -44,16 +45,10 @@
void removeConsumerRegistrationListener(ConsumerRegistrationListener<S> listener);
- AuthorizationHolder getAuthorizationHolder();
-
- void setAuthorizationHolder(AuthorizationHolder principalHolder);
-
- void setExclusiveOwningSession(AMQSessionModel owner);
-
- AMQSessionModel getExclusiveOwningSession();
-
boolean isExclusive();
+ boolean verifySessionAccess(AMQSessionModel<?,?> session);
+
interface ConsumerRegistrationListener<Q extends MessageSource<? extends Consumer,?>>
{
void consumerAdded(Q source, Consumer consumer);
@@ -76,7 +71,6 @@
public ExistingExclusiveConsumer()
{
- super("");
}
}
@@ -95,7 +89,15 @@
{
public ExistingConsumerPreventsExclusive()
{
- super("");
}
}
+
+ static final class ConsumerAccessRefused extends Exception
+ {
+ public ConsumerAccessRefused()
+ {
+ }
+ }
+
+
}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/model/ExclusivityPolicy.java b/java/broker-core/src/main/java/org/apache/qpid/server/model/ExclusivityPolicy.java
new file mode 100644
index 0000000..a559a69
--- /dev/null
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/model/ExclusivityPolicy.java
@@ -0,0 +1,31 @@
+/*
+ *
+ * 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.model;
+
+public enum ExclusivityPolicy
+{
+ NONE,
+ SESSION,
+ CONNECTION,
+ CONTAINER,
+ PRINCIPAL,
+ LINK
+}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/model/LifetimePolicy.java b/java/broker-core/src/main/java/org/apache/qpid/server/model/LifetimePolicy.java
index c9006f4..1e45f8f 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/model/LifetimePolicy.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/model/LifetimePolicy.java
@@ -23,5 +23,9 @@
public enum LifetimePolicy
{
PERMANENT,
- AUTO_DELETE
+ DELETE_ON_CONNECTION_CLOSE,
+ DELETE_ON_SESSION_END,
+ DELETE_ON_NO_OUTBOUND_LINKS,
+ DELETE_ON_NO_LINKS,
+ IN_USE
}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java b/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
index f0241f8..dd3e9e4 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
@@ -160,7 +160,7 @@
QUEUE_ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES,
CONFIG_PATH));
- int CURRENT_CONFIG_VERSION = 3;
+ int CURRENT_CONFIG_VERSION = 4;
//children
Collection<VirtualHostAlias> getAliases();
@@ -172,9 +172,8 @@
LifetimePolicy lifetime, long ttl, String type, Map<String, Object> attributes)
throws AccessControlException, IllegalArgumentException;
- Queue createQueue(String name, State initialState, boolean durable,
- boolean exclusive, LifetimePolicy lifetime, long ttl, Map<String, Object> attributes)
- throws AccessControlException, IllegalArgumentException;
+ Queue createQueue(Map<String, Object> attributes)
+ throws AccessControlException, IllegalArgumentException;
Collection<String> getExchangeTypes();
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/BindingAdapter.java b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/BindingAdapter.java
index 8847580..265d431 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/BindingAdapter.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/BindingAdapter.java
@@ -170,7 +170,7 @@
}
else if(LIFETIME_POLICY.equals(name))
{
- return _queue.getLifetimePolicy() == LifetimePolicy.AUTO_DELETE || _exchange.getLifetimePolicy() == LifetimePolicy.AUTO_DELETE ? LifetimePolicy.AUTO_DELETE : LifetimePolicy.PERMANENT;
+ return _queue.getLifetimePolicy() != LifetimePolicy.PERMANENT || _exchange.getLifetimePolicy() != LifetimePolicy.PERMANENT ? LifetimePolicy.IN_USE : LifetimePolicy.PERMANENT;
}
else if(TIME_TO_LIVE.equals(name))
{
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConnectionAdapter.java b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConnectionAdapter.java
index dda23f1..3cb6493 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConnectionAdapter.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConnectionAdapter.java
@@ -21,6 +21,7 @@
package org.apache.qpid.server.model.adapter;
import java.security.AccessControlException;
+import java.security.Principal;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -191,7 +192,8 @@
}
else if(name.equals(PRINCIPAL))
{
- return _connection.getPrincipalAsString();
+ final Principal authorizedPrincipal = _connection.getAuthorizedPrincipal();
+ return authorizedPrincipal == null ? null : authorizedPrincipal.getName();
}
else if(name.equals(PROPERTIES))
{
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java
index cf68740..7935077 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java
@@ -130,7 +130,7 @@
}
else if(LIFETIME_POLICY.equals(name))
{
- return LifetimePolicy.AUTO_DELETE;
+ return LifetimePolicy.DELETE_ON_SESSION_END;
}
else if(TIME_TO_LIVE.equals(name))
{
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ExchangeAdapter.java b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ExchangeAdapter.java
index f7ecbd3..d7b6b8b 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ExchangeAdapter.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ExchangeAdapter.java
@@ -201,7 +201,7 @@
public LifetimePolicy getLifetimePolicy()
{
- return _exchange.isAutoDelete() ? LifetimePolicy.AUTO_DELETE : LifetimePolicy.PERMANENT;
+ return _exchange.isAutoDelete() ? LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS : LifetimePolicy.PERMANENT;
}
public LifetimePolicy setLifetimePolicy(final LifetimePolicy expected, final LifetimePolicy desired)
@@ -330,7 +330,7 @@
}
else if(LIFETIME_POLICY.equals(name))
{
- return _exchange.isAutoDelete() ? LifetimePolicy.AUTO_DELETE : LifetimePolicy.PERMANENT;
+ return _exchange.isAutoDelete() ? LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS : LifetimePolicy.PERMANENT;
}
else if(TIME_TO_LIVE.equals(name))
{
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java
index 733b6ce..5d09cfa 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java
@@ -31,15 +31,7 @@
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.configuration.IllegalConfigurationException;
import org.apache.qpid.server.message.MessageSource;
-import org.apache.qpid.server.model.ConfiguredObject;
-import org.apache.qpid.server.model.ConfiguredObjectFinder;
-import org.apache.qpid.server.model.Exchange;
-import org.apache.qpid.server.model.IllegalStateTransitionException;
-import org.apache.qpid.server.model.LifetimePolicy;
-import org.apache.qpid.server.model.Queue;
-import org.apache.qpid.server.model.QueueNotificationListener;
-import org.apache.qpid.server.model.State;
-import org.apache.qpid.server.model.Statistics;
+import org.apache.qpid.server.model.*;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.queue.*;
@@ -62,7 +54,6 @@
put(QUEUE_FLOW_CONTROL_SIZE_BYTES, Long.class);
put(QUEUE_FLOW_RESUME_SIZE_BYTES, Long.class);
put(MAXIMUM_DELIVERY_ATTEMPTS, Integer.class);
- put(EXCLUSIVE, Boolean.class);
put(DESCRIPTION, String.class);
}});
@@ -208,7 +199,7 @@
public LifetimePolicy getLifetimePolicy()
{
- return _queue.isAutoDelete() ? LifetimePolicy.AUTO_DELETE : LifetimePolicy.PERMANENT;
+ return _queue.getLifetimePolicy();
}
public LifetimePolicy setLifetimePolicy(final LifetimePolicy expected, final LifetimePolicy desired)
@@ -274,9 +265,33 @@
}
else if(EXCLUSIVE.equals(name))
{
- Boolean exclusiveFlag = (Boolean) desired;
- _queue.setExclusive(exclusiveFlag);
+ ExclusivityPolicy desiredPolicy;
+ if(desired == null)
+ {
+ desiredPolicy = ExclusivityPolicy.NONE;
+ }
+ else if(desired instanceof ExclusivityPolicy)
+ {
+ desiredPolicy = (ExclusivityPolicy)desired;
+ }
+ else if (desired instanceof String)
+ {
+ desiredPolicy = ExclusivityPolicy.valueOf((String)desired);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Cannot set " + Queue.EXCLUSIVE + " property to type " + desired.getClass().getName());
+ }
+ try
+ {
+ _queue.setExclusivityPolicy(desiredPolicy);
+ }
+ catch (MessageSource.ExistingConsumerPreventsExclusive existingConsumerPreventsExclusive)
+ {
+ throw new IllegalArgumentException("Unable to set exclusivity policy to " + desired + " as an existing combinations of consumers prevents this");
+ }
return true;
+
}
else if(MESSAGE_GROUP_KEY.equals(name))
{
@@ -376,7 +391,7 @@
}
else if(EXCLUSIVE.equals(name))
{
- return _queue.isExclusive();
+ return _queue.getAttribute(Queue.EXCLUSIVE);
}
else if(MESSAGE_GROUP_KEY.equals(name))
{
@@ -458,7 +473,7 @@
}
else if(LIFETIME_POLICY.equals(name))
{
- return _queue.isAutoDelete() ? LifetimePolicy.AUTO_DELETE : LifetimePolicy.PERMANENT;
+ return _queue.getLifetimePolicy();
}
else if(NAME.equals(name))
{
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java
index 17ce69d..4cd7432 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java
@@ -23,7 +23,6 @@
import java.io.File;
import java.lang.reflect.Type;
import java.security.AccessControlException;
-import java.security.Principal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -58,18 +57,16 @@
import org.apache.qpid.server.model.QueueType;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.Statistics;
-import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.model.VirtualHostAlias;
import org.apache.qpid.server.configuration.updater.TaskExecutor;
import org.apache.qpid.server.plugin.ExchangeType;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.AMQQueueFactory;
+import org.apache.qpid.server.queue.ConflationQueue;
import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.security.access.Operation;
-import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
import org.apache.qpid.server.stats.StatisticsGatherer;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.txn.LocalTransaction;
@@ -356,7 +353,7 @@
name,
type,
durable,
- lifetime == LifetimePolicy.AUTO_DELETE,
+ lifetime != null && lifetime != LifetimePolicy.PERMANENT,
alternateExchange);
synchronized (_exchangeAdapters)
{
@@ -389,7 +386,7 @@
public Queue createQueue(Map<String, Object> attributes)
throws AccessControlException, IllegalArgumentException
{
- attributes = new HashMap<String, Object>(attributes);
+ checkVHostStateIsActive();
if (attributes.containsKey(Queue.QUEUE_TYPE))
{
@@ -405,7 +402,7 @@
}
if (queueType == QueueType.LVQ && attributes.get(Queue.LVQ_KEY) == null)
{
- attributes.put(Queue.LVQ_KEY, AMQQueueFactory.QPID_DEFAULT_LVQ_KEY);
+ attributes.put(Queue.LVQ_KEY, ConflationQueue.DEFAULT_LVQ_KEY);
}
else if (queueType == QueueType.PRIORITY && attributes.get(Queue.PRIORITIES) == null)
{
@@ -417,51 +414,12 @@
}
}
- String name = MapValueConverter.getStringAttribute(Queue.NAME, attributes, null);
- State state = MapValueConverter.getEnumAttribute(State.class, Queue.STATE, attributes, State.ACTIVE);
- boolean durable = MapValueConverter.getBooleanAttribute(Queue.DURABLE, attributes, false);
- LifetimePolicy lifetime = MapValueConverter.getEnumAttribute(LifetimePolicy.class, Queue.LIFETIME_POLICY, attributes, LifetimePolicy.PERMANENT);
- long ttl = MapValueConverter.getLongAttribute(Queue.TIME_TO_LIVE, attributes, 0l);
- boolean exclusive= MapValueConverter.getBooleanAttribute(Queue.EXCLUSIVE, attributes, false);
- attributes.remove(Queue.NAME);
- attributes.remove(Queue.STATE);
- attributes.remove(Queue.DURABLE);
- attributes.remove(Queue.LIFETIME_POLICY);
- attributes.remove(Queue.TIME_TO_LIVE);
-
- return createQueue(name, state, durable, exclusive, lifetime, ttl, attributes);
- }
-
- public Queue createQueue(final String name,
- final State initialState,
- final boolean durable,
- boolean exclusive,
- final LifetimePolicy lifetime,
- final long ttl,
- final Map<String, Object> attributes)
- throws AccessControlException, IllegalArgumentException
- {
- checkVHostStateIsActive();
-
- String owner = null;
- if(exclusive)
- {
- Principal authenticatedPrincipal = AuthenticatedPrincipal.getOptionalAuthenticatedPrincipalFromSubject(SecurityManager.getThreadSubject());
- if(authenticatedPrincipal != null)
- {
- owner = authenticatedPrincipal.getName();
- }
- }
-
- final boolean autoDelete = lifetime == LifetimePolicy.AUTO_DELETE;
try
{
- AMQQueue queue =
- _virtualHost.createQueue(UUIDGenerator.generateQueueUUID(name, _virtualHost.getName()), name,
- durable, owner, autoDelete, exclusive, autoDelete && exclusive, attributes);
+ AMQQueue queue = _virtualHost.createQueue(null, attributes);
synchronized (_queueAdapters)
{
@@ -471,15 +429,15 @@
}
catch(QueueExistsException qe)
{
- throw new IllegalArgumentException("Queue with name "+name+" already exists");
+ throw new IllegalArgumentException("Queue with name "+MapValueConverter.getStringAttribute(Queue.NAME,attributes)+" already exists");
}
catch (QpidSecurityException e)
{
throw new AccessControlException(e.toString());
}
-
}
+
public String getName()
{
return (String)getAttribute(NAME);
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java b/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java
index bc272f1..623cf62 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java
@@ -25,11 +25,13 @@
import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.stats.StatisticsGatherer;
+import org.apache.qpid.server.util.Deletable;
+import java.security.Principal;
import java.util.List;
import java.util.UUID;
-public interface AMQConnectionModel extends StatisticsGatherer
+public interface AMQConnectionModel<T extends AMQConnectionModel<T,S>, S extends AMQSessionModel<S,T>> extends StatisticsGatherer, Deletable<T>
{
/**
* Close the underlying Connection
@@ -50,7 +52,7 @@
* @param cause
* @param message
*/
- public void closeSession(AMQSessionModel session, AMQConstant cause, String message);
+ public void closeSession(S session, AMQConstant cause, String message);
public long getConnectionId();
@@ -59,15 +61,13 @@
*
* @return a list of {@link AMQSessionModel}s
*/
- public List<AMQSessionModel> getSessionModels();
+ public List<S> getSessionModels();
/**
* Return a {@link LogSubject} for the connection.
*/
public LogSubject getLogSubject();
- public String getUserName();
-
public boolean isSessionNameUnique(byte[] name);
String getRemoteAddressString();
@@ -78,7 +78,7 @@
String getClientProduct();
- String getPrincipalAsString();
+ Principal getAuthorizedPrincipal();
long getSessionCountLimit();
@@ -93,4 +93,6 @@
boolean isStopped();
String getVirtualHostName();
+
+ String getRemoteContainerName();
}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java b/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
index 520489a..db92f69 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
@@ -26,17 +26,18 @@
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.util.Deletable;
/**
* Session model interface.
* Extends {@link Comparable} to allow objects to be inserted into a {@link ConcurrentSkipListSet}
* when monitoring the blocking and blocking of queues/sessions in {@link AMQQueue}.
*/
-public interface AMQSessionModel extends Comparable<AMQSessionModel>
+public interface AMQSessionModel<T extends AMQSessionModel<T,C>, C extends AMQConnectionModel<C,T>> extends Comparable<T>, Deletable<T>
{
public UUID getId();
- public AMQConnectionModel getConnectionModel();
+ public C getConnectionModel();
public String getClientID();
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
index 237f2cd..aceebc7 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
@@ -27,10 +27,13 @@
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.message.MessageDestination;
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.CapacityChecker;
import org.apache.qpid.server.consumer.Consumer;
import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.server.util.Action;
+import org.apache.qpid.server.util.Deletable;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Collection;
@@ -38,9 +41,12 @@
import java.util.Set;
public interface AMQQueue<E extends QueueEntry<E,Q,C>, Q extends AMQQueue<E,Q,C>, C extends Consumer>
- extends Comparable<Q>, ExchangeReferrer, BaseQueue<C>, MessageSource<C,Q>, CapacityChecker, MessageDestination
+ extends Comparable<Q>, ExchangeReferrer, BaseQueue<C>, MessageSource<C,Q>, CapacityChecker, MessageDestination,
+ Deletable<Q>
{
+ void setExclusivityPolicy(ExclusivityPolicy desiredPolicy) throws ExistingConsumerPreventsExclusive;
+
public interface NotificationListener
{
void notifyClients(NotificationCheck notification, AMQQueue queue, String notificationMsg);
@@ -66,9 +72,7 @@
long getTotalEnqueueCount();
- void setNoLocal(boolean b);
-
- boolean isAutoDelete();
+ LifetimePolicy getLifetimePolicy();
String getOwner();
@@ -104,11 +108,6 @@
boolean resend(final E entry, final C consumer);
- void addQueueDeleteTask(Action<AMQQueue> task);
- void removeQueueDeleteTask(Action<AMQQueue> task);
-
-
-
List<E> getMessagesOnTheQueue();
List<Long> getMessagesOnTheQueue(int num);
@@ -189,10 +188,6 @@
Collection<String> getAvailableAttributes();
Object getAttribute(String attrName);
- void configure(QueueConfiguration config);
-
- void setExclusive(boolean exclusive);
-
/**
* Gets the maximum delivery count. If a message on this queue
* is delivered more than maximumDeliveryCount, the message will be
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java b/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java
index 409c528..5003db1 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java
@@ -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.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 @@
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,346 +61,101 @@
{
_virtualHost = virtualHost;
_queueRegistry = queueRegistry;
- }
-
- private abstract static class QueueProperty
- {
-
- private final String _argumentName;
-
-
- public QueueProperty(String argumentName)
- {
- _argumentName = argumentName;
- }
-
- public String getArgumentName()
- {
- return _argumentName;
- }
-
-
- public abstract void setPropertyValue(AMQQueue queue, Object value);
-
- }
-
- private abstract static class QueueLongProperty extends QueueProperty
- {
-
- public QueueLongProperty(String argumentName)
- {
- super(argumentName);
- }
-
- public void setPropertyValue(AMQQueue queue, Object value)
- {
- if(value instanceof Number)
- {
- setPropertyValue(queue, ((Number)value).longValue());
- }
-
- }
-
- abstract void setPropertyValue(AMQQueue queue, long value);
-
-
- }
-
- private abstract static class QueueIntegerProperty extends QueueProperty
- {
- public QueueIntegerProperty(String argumentName)
- {
- super(argumentName);
- }
-
- public void setPropertyValue(AMQQueue queue, Object value)
- {
- if(value instanceof Number)
- {
- setPropertyValue(queue, ((Number)value).intValue());
- }
-
- }
- 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
+ public AMQQueue restoreQueue(Map<String, Object> attributes) throws QpidSecurityException
{
- return createOrRestoreQueue(id, queueName, true, owner, autoDelete, exclusive, deleteOnNoConsumer, arguments, false);
+ return createOrRestoreQueue(null, attributes, 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
+ public AMQQueue createQueue(final AMQSessionModel creatingSession,
+ Map<String, Object> attributes) throws QpidSecurityException
{
- return createOrRestoreQueue(id, queueName, durable, owner, autoDelete, exclusive, deleteOnNoConsumer, arguments, true);
+ return createOrRestoreQueue(creatingSession, attributes, true);
}
- private AMQQueue createOrRestoreQueue(UUID id,
- String queueName,
- boolean durable,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- boolean deleteOnNoConsumer,
- Map<String, Object> arguments,
+ private AMQQueue createOrRestoreQueue(final AMQSessionModel creatingSession, Map<String, Object> attributes,
boolean createInStore) throws QpidSecurityException
{
- if (id == null)
+
+
+ String queueName = MapValueConverter.getStringAttribute(Queue.NAME,attributes);
+
+ QueueConfiguration config = _virtualHost.getConfiguration().getQueueConfiguration(queueName);
+
+ if (!attributes.containsKey(Queue.ALERT_THRESHOLD_MESSAGE_AGE) && config.getMaximumMessageAge() != 0)
{
- throw new IllegalArgumentException("Queue id must not be null");
+ attributes.put(Queue.ALERT_THRESHOLD_MESSAGE_AGE, config.getMaximumMessageAge());
}
- if (queueName == null)
+ if (!attributes.containsKey(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_BYTES) && config.getMaximumQueueDepth() != 0)
{
- throw new IllegalArgumentException("Queue name must not be null");
+ attributes.put(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_BYTES, config.getMaximumQueueDepth());
+ }
+ if (!attributes.containsKey(Queue.ALERT_THRESHOLD_MESSAGE_SIZE) && config.getMaximumMessageSize() != 0)
+ {
+ attributes.put(Queue.ALERT_THRESHOLD_MESSAGE_SIZE, config.getMaximumMessageSize());
+ }
+ if (!attributes.containsKey(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES) && config.getMaximumMessageCount() != 0)
+ {
+ attributes.put(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES, config.getMaximumMessageCount());
+ }
+ if (!attributes.containsKey(Queue.ALERT_REPEAT_GAP) && config.getMinimumAlertRepeatGap() != 0)
+ {
+ attributes.put(Queue.ALERT_REPEAT_GAP, config.getMinimumAlertRepeatGap());
+ }
+ if (config.getMaxDeliveryCount() != 0 && !attributes.containsKey(Queue.MAXIMUM_DELIVERY_ATTEMPTS))
+ {
+ attributes.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, config.getMaxDeliveryCount());
+ }
+ if (!attributes.containsKey(Queue.QUEUE_FLOW_CONTROL_SIZE_BYTES) && config.getCapacity() != 0)
+ {
+ attributes.put(Queue.QUEUE_FLOW_CONTROL_SIZE_BYTES, config.getCapacity());
+ }
+ if (!attributes.containsKey(Queue.QUEUE_FLOW_RESUME_SIZE_BYTES) && config.getFlowResumeCapacity() != 0)
+ {
+ 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;
+ AMQQueue queue;
- if(arguments != null)
+ if(attributes.containsKey(Queue.SORT_KEY))
{
- 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);
- }
+ queue = new SortedQueue(_virtualHost, creatingSession, attributes);
}
-
- AMQQueue q;
- if(sortingKey != null)
+ else if(attributes.containsKey(Queue.LVQ_KEY))
{
- q = new SortedQueue(id, queueName, durable, owner, autoDelete, exclusive, _virtualHost, arguments, sortingKey);
+ queue = new ConflationQueue(_virtualHost, creatingSession, attributes);
}
- else if(conflationKey != null)
+ else if(attributes.containsKey(Queue.PRIORITIES))
{
- q = new ConflationQueue(id, queueName, durable, owner, autoDelete, exclusive, _virtualHost, arguments, conflationKey);
- }
- else if(priorities > 1)
- {
- 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);
-
- q.configure(_virtualHost.getConfiguration().getQueueConfiguration(queueName));
-
- if(arguments != null)
- {
- 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));
- }
-
- }
+ _queueRegistry.registerQueue(queue);
if(createDLQ)
{
- final String dlExchangeName = getDeadLetterExchangeName(queueName);
- final String dlQueueName = getDeadLetterQueueName(queueName);
-
- 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);
- }
- 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);
- }
-
- AMQQueue dlQueue = null;
-
- synchronized(_queueRegistry)
- {
- dlQueue = _queueRegistry.getQueue(dlQueueName);
-
- 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);
- }
- }
- }
-
- //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);
+ createDLQ(queue);
}
- else if(arguments != null && arguments.get(Queue.ALTERNATE_EXCHANGE) instanceof String)
+ else if(attributes != null && attributes.get(Queue.ALTERNATE_EXCHANGE) instanceof String)
{
- final String altExchangeAttr = (String) arguments.get(Queue.ALTERNATE_EXCHANGE);
+ final String altExchangeAttr = (String) attributes.get(Queue.ALTERNATE_EXCHANGE);
Exchange altExchange;
try
{
@@ -408,32 +165,103 @@
{
altExchange = _virtualHost.getExchange(altExchangeAttr);
}
- q.setAlternateExchange(altExchange);
+ queue.setAlternateExchange(altExchange);
}
- if (createInStore && q.isDurable() && !q.isAutoDelete())
+ if (createInStore && queue.isDurable() && !(queue.getLifetimePolicy()
+ == LifetimePolicy.DELETE_ON_CONNECTION_CLOSE
+ || queue.getLifetimePolicy()
+ == LifetimePolicy.DELETE_ON_SESSION_END))
{
- DurableConfigurationStoreHelper.createQueue(_virtualHost.getDurableConfigurationStore(), q);
+ DurableConfigurationStoreHelper.createQueue(_virtualHost.getDurableConfigurationStore(), queue);
}
- return q;
+ return queue;
+ }
+
+ private void createDLQ(final AMQQueue queue) throws QpidSecurityException
+ {
+ final String queueName = queue.getName();
+ final String dlExchangeName = getDeadLetterExchangeName(queueName);
+ final String dlQueueName = getDeadLetterQueueName(queueName);
+
+ 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);
+ }
+ 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);
+ }
+
+ AMQQueue dlQueue = null;
+
+ synchronized(_queueRegistry)
+ {
+ dlQueue = _queueRegistry.getQueue(dlQueueName);
+
+ 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
+ {
+
+
+ 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);
+ }
+ }
+ }
+
+ //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);
+ }
+ 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 @@
/**
* 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 @@
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;
}
}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueue.java b/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueue.java
index a1ff519..f350368 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueue.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueue.java
@@ -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()
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java b/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java
index 7cf245c..c9ac3f3 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java
@@ -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
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueue.java b/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueue.java
index 4440d04..1e41a0f 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueue.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueue.java
@@ -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(id, name, durable, owner, autoDelete, exclusive, virtualHost, new PriorityQueueList.Factory(priorities), arguments);
+ super(virtualHost, creatingSession, attributes, entryList(attributes));
+ }
+
+ private static PriorityQueueList.Factory entryList(final Map<String, Object> attributes)
+ {
+ final Integer priorities = MapValueConverter.getIntegerAttribute(Queue.PRIORITIES, attributes,
+ DEFAULT_PRIORITY_LEVELS);
+
+ return new PriorityQueueList.Factory(priorities);
}
public int getPriorities()
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueArgumentsConverter.java b/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueArgumentsConverter.java
index 589f385..49123f8 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueArgumentsConverter.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueArgumentsConverter.java
@@ -109,7 +109,7 @@
}
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))
{
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueFactory.java b/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueFactory.java
index 8c0386c..62a2d93 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueFactory.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueFactory.java
@@ -22,25 +22,15 @@
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;
}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
index b2408f6..aa7025e 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
@@ -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.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.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 @@
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 @@
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 @@
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 @@
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 @@
_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 @@
_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 @@
}
+ 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 @@
public boolean isExclusive()
{
- return _exclusive;
- }
-
- public void setExclusive(boolean exclusive)
- {
- _exclusive = exclusive;
+ return _exclusivityPolicy != ExclusivityPolicy.NONE;
}
public Exchange getAlternateExchange()
@@ -342,27 +544,27 @@
return _arguments.get(attrName);
}
- public boolean isAutoDelete()
+ @Override
+ public LifetimePolicy getLifetimePolicy()
{
- return _autoDelete;
+ return _lifetimePolicy;
}
public String getOwner()
{
- return _owner;
+ if(_exclusiveOwner != null)
+ {
+ switch(_exclusivityPolicy)
+ {
+ case CONTAINER:
+ return (String) _exclusiveOwner;
+ case PRINCIPAL:
+ return ((Principal)_exclusiveOwner).getName();
+ }
+ }
+ return null;
}
- public AuthorizationHolder getAuthorizationHolder()
- {
- return _authorizationHolder;
- }
-
- public void setAuthorizationHolder(final AuthorizationHolder authorizationHolder)
- {
- _authorizationHolder = authorizationHolder;
- }
-
-
public VirtualHost getVirtualHost()
{
return _virtualHost;
@@ -381,7 +583,9 @@
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 @@
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 @@
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 @@
// 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 @@
});
}
- 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 @@
}
- for (Action<AMQQueue> task : _deleteTaskList)
+ for (Action<? super Q> task : _deleteTaskList)
{
- task.performAction(this);
+ task.performAction((Q)this);
}
_deleteTaskList.clear();
@@ -1940,6 +2213,26 @@
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 @@
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 @@
@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 @@
}
+ @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;
+ }
+ }
}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java b/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java
index f85f947..d1da7fe 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java
@@ -21,11 +21,13 @@
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 @@
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;
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/queue/StandardQueue.java b/java/broker-core/src/main/java/org/apache/qpid/server/queue/StandardQueue.java
index e2ccdc4..8a0570e 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/queue/StandardQueue.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/queue/StandardQueue.java
@@ -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());
}
}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java b/java/broker-core/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java
index a379f85..b9e0e59 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java
@@ -27,6 +27,7 @@
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 @@
{
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 @@
{
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)
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java b/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java
index fc5806c..096734f 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java
@@ -42,17 +42,11 @@
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 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 @@
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);
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/util/Deletable.java b/java/broker-core/src/main/java/org/apache/qpid/server/util/Deletable.java
new file mode 100644
index 0000000..a6b16bb
--- /dev/null
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/util/Deletable.java
@@ -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);
+}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/util/MapValueConverter.java b/java/broker-core/src/main/java/org/apache/qpid/server/util/MapValueConverter.java
index 37e0177..3543ce3 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/util/MapValueConverter.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/util/MapValueConverter.java
@@ -28,6 +28,7 @@
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
public class MapValueConverter
{
@@ -217,6 +218,13 @@
return getIntegerAttribute(name, attributes, null);
}
+ public static Long getLongAttribute(String name, Map<String,Object> attributes)
+ {
+ assertMandatoryAttribute(name, attributes);
+ Object obj = attributes.get(name);
+ return toLong(name, obj, null);
+ }
+
public static Long getLongAttribute(String name, Map<String,Object> attributes, Long defaultValue)
{
Object obj = attributes.get(name);
@@ -409,4 +417,41 @@
return (T) value;
}
+
+ public static UUID getUUIDAttribute(String name, Map<String, Object> attributes)
+ {
+ assertMandatoryAttribute(name, attributes);
+ return getUUIDAttribute(name, attributes, null);
+ }
+
+ public static UUID getUUIDAttribute(String name, Map<String,Object> attributes, UUID defaultVal)
+ {
+ final Object value = attributes.get(name);
+ return toUUID(value, defaultVal);
+ }
+
+ private static UUID toUUID(final Object value, final UUID defaultVal)
+ {
+ if(value == null)
+ {
+ return defaultVal;
+ }
+ else if(value instanceof UUID)
+ {
+ return (UUID)value;
+ }
+ else if(value instanceof String)
+ {
+ return UUID.fromString((String)value);
+ }
+ else if(value instanceof byte[])
+ {
+ return UUID.nameUUIDFromBytes((byte[])value);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Cannot convert " + value.getClass().getName() + " to UUID");
+ }
+ }
+
}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java b/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
index aa70bb3..0e9b879 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
@@ -23,6 +23,7 @@
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -34,6 +35,9 @@
import org.apache.commons.configuration.ConfigurationException;
import org.apache.log4j.Logger;
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.model.Queue;
import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.server.configuration.ExchangeConfiguration;
import org.apache.qpid.server.configuration.QueueConfiguration;
@@ -69,7 +73,9 @@
import org.apache.qpid.server.store.DurableConfiguredObjectRecoverer;
import org.apache.qpid.server.store.Event;
import org.apache.qpid.server.store.EventListener;
+import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.txn.DtxRegistry;
+import org.apache.qpid.server.util.MapValueConverter;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
public abstract class AbstractVirtualHost implements VirtualHost, IConnectionRegistry.RegistryChangeListener, EventListener
@@ -529,7 +535,10 @@
int purged = queue.delete();
getQueueRegistry().unregisterQueue(queue.getName());
- if (queue.isDurable() && !queue.isAutoDelete())
+ if (queue.isDurable() && !(queue.getLifetimePolicy()
+ == LifetimePolicy.DELETE_ON_CONNECTION_CLOSE
+ || queue.getLifetimePolicy()
+ == LifetimePolicy.DELETE_ON_SESSION_END))
{
DurableConfigurationStore store = getDurableConfigurationStore();
DurableConfigurationStoreHelper.removeQueue(store, queue);
@@ -538,26 +547,24 @@
}
}
- @Override
- public AMQQueue createQueue(UUID id,
- String queueName,
- boolean durable,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- boolean deleteOnNoConsumer,
- Map<String, Object> arguments) throws QpidSecurityException, QueueExistsException
+ public AMQQueue createQueue(final AMQSessionModel creatingSession, Map<String, Object> attributes) throws QpidSecurityException, QueueExistsException
{
+ // make a copy as we may augment (with an ID for example)
+ attributes = new LinkedHashMap<String, Object>(attributes);
- if (queueName == null)
- {
- throw new IllegalArgumentException("Queue name must not be null");
- }
+ String queueName = MapValueConverter.getStringAttribute(Queue.NAME, attributes);
+ boolean autoDelete = MapValueConverter.getEnumAttribute(LifetimePolicy.class,
+ Queue.LIFETIME_POLICY,
+ attributes,
+ LifetimePolicy.PERMANENT) != LifetimePolicy.PERMANENT;
+ boolean durable = MapValueConverter.getBooleanAttribute(Queue.DURABLE, attributes, false);
+ ExclusivityPolicy exclusive = MapValueConverter.getEnumAttribute(ExclusivityPolicy.class,Queue.EXCLUSIVE, attributes, ExclusivityPolicy.NONE);
+ String owner = MapValueConverter.getStringAttribute(Queue.OWNER, attributes, null);
- // Access check
+ // Access check
if (!getSecurityManager().authoriseCreateQueue(autoDelete,
durable,
- exclusive,
+ exclusive != null && exclusive != ExclusivityPolicy.NONE,
null,
null,
queueName,
@@ -573,22 +580,27 @@
{
throw new QueueExistsException("Queue with name " + queueName + " already exists", _queueRegistry.getQueue(queueName));
}
- if(id == null)
+ if(!attributes.containsKey(Queue.ID))
{
- id = UUIDGenerator.generateExchangeUUID(queueName, getName());
+ UUID id = UUIDGenerator.generateExchangeUUID(queueName, getName());
while(_queueRegistry.getQueue(id) != null)
{
id = UUID.randomUUID();
}
+ attributes.put(Queue.ID, id);
}
- else if(_queueRegistry.getQueue(id) != null)
+ else if(_queueRegistry.getQueue(MapValueConverter.getUUIDAttribute(Queue.ID, attributes)) != null)
{
- throw new QueueExistsException("Queue with id " + id + " already exists", _queueRegistry.getQueue(queueName));
+ throw new QueueExistsException("Queue with id "
+ + MapValueConverter.getUUIDAttribute(Queue.ID,
+ attributes)
+ + " already exists", _queueRegistry.getQueue(queueName));
}
- return _queueFactory.createQueue(id, queueName, durable, owner, autoDelete, exclusive, deleteOnNoConsumer,
- arguments);
+
+
+ return _queueFactory.createQueue(creatingSession, attributes);
}
}
@@ -980,13 +992,13 @@
// house keeping task from running.
}
}
- for (AMQConnectionModel connection : getConnectionRegistry().getConnections())
+ for (AMQConnectionModel<?,?> connection : getConnectionRegistry().getConnections())
{
if (_logger.isDebugEnabled())
{
_logger.debug("Checking for long running open transactions on connection " + connection);
}
- for (AMQSessionModel session : connection.getSessionModels())
+ for (AMQSessionModel<?,?> session : connection.getSessionModels())
{
if (_logger.isDebugEnabled())
{
@@ -1046,5 +1058,54 @@
{
return _model;
}
+
+ }
+
+ @Override
+ public long getDefaultAlertThresholdMessageAge()
+ {
+ return getConfiguration().getMaximumMessageAge();
+ }
+
+ @Override
+ public long getDefaultAlertThresholdMessageSize()
+ {
+ return getConfiguration().getMaximumMessageSize();
+ }
+
+ @Override
+ public long getDefaultAlertThresholdQueueDepthMessages()
+ {
+ return getConfiguration().getMaximumMessageCount();
+ }
+
+ @Override
+ public long getDefaultAlertThresholdQueueDepthBytes()
+ {
+ return getConfiguration().getMaximumQueueDepth();
+ }
+
+ @Override
+ public long getDefaultAlertRepeatGap()
+ {
+ return getConfiguration().getMinimumAlertRepeatGap();
+ }
+
+ @Override
+ public long getDefaultQueueFlowControlSizeBytes()
+ {
+ return getConfiguration().getCapacity();
+ }
+
+ @Override
+ public long getDefaultQueueFlowResumeSizeBytes()
+ {
+ return getConfiguration().getFlowResumeCapacity();
+ }
+
+ @Override
+ public int getDefaultMaximumDeliveryAttempts()
+ {
+ return getConfiguration().getMaxDeliveryCount();
}
}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/DefaultUpgraderProvider.java b/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/DefaultUpgraderProvider.java
index 12f8c7d..88caf73 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/DefaultUpgraderProvider.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/DefaultUpgraderProvider.java
@@ -42,6 +42,7 @@
public class DefaultUpgraderProvider implements UpgraderProvider
{
+ public static final String EXCLUSIVE = "exclusive";
private final ExchangeRegistry _exchangeRegistry;
private final VirtualHost _virtualHost;
@@ -63,7 +64,8 @@
currentUpgrader = addUpgrader(currentUpgrader, new Version1Upgrader());
case 2:
currentUpgrader = addUpgrader(currentUpgrader, new Version2Upgrader());
-
+ case 3:
+ currentUpgrader = addUpgrader(currentUpgrader, new Version3Upgrader());
case CURRENT_CONFIG_VERSION:
currentUpgrader = addUpgrader(currentUpgrader, new NullUpgrader(recoverer));
break;
@@ -263,4 +265,49 @@
}
}
+ /*
+ * Convert the storage of queue attribute exclusive to change exclusive from a boolean to an enum
+ * where exclusive was false it will now be "NONE", and where true it will now be "CONTAINER"
+ * ensure OWNER is null unless the exclusivity policy is CONTAINER
+ */
+ private class Version3Upgrader extends NonNullUpgrader
+ {
+
+ @Override
+ public void configuredObject(UUID id, String type, Map<String, Object> attributes)
+ {
+ if(Queue.class.getSimpleName().equals(type))
+ {
+ Map<String, Object> newAttributes = new LinkedHashMap<String, Object>(attributes);
+ if(attributes.get(EXCLUSIVE) instanceof Boolean)
+ {
+ boolean isExclusive = (Boolean) attributes.get(EXCLUSIVE);
+ newAttributes.put(EXCLUSIVE, isExclusive ? "CONTAINER" : "NONE");
+ if(!isExclusive && attributes.containsKey("owner"))
+ {
+ newAttributes.remove("owner");
+ }
+ }
+ else
+ {
+ newAttributes.remove("owner");
+ }
+ if(!attributes.containsKey("durable"))
+ {
+ newAttributes.put("durable","true");
+ }
+ attributes = newAttributes;
+ getUpdateMap().put(id, new ConfiguredObjectRecord(id,type,attributes));
+ }
+
+ getNextUpgrader().configuredObject(id,type,attributes);
+ }
+
+ @Override
+ public void complete()
+ {
+ getNextUpgrader().complete();
+ }
+ }
+
}
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ExchangeRecoverer.java b/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ExchangeRecoverer.java
index 59ff1ce..c687cbd 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ExchangeRecoverer.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ExchangeRecoverer.java
@@ -69,7 +69,7 @@
String exchangeType = (String) attributeMap.get(org.apache.qpid.server.model.Exchange.TYPE);
String lifeTimePolicy = (String) attributeMap.get(org.apache.qpid.server.model.Exchange.LIFETIME_POLICY);
boolean autoDelete = lifeTimePolicy == null
- || LifetimePolicy.valueOf(lifeTimePolicy) == LifetimePolicy.AUTO_DELETE;
+ || LifetimePolicy.valueOf(lifeTimePolicy) != LifetimePolicy.PERMANENT;
try
{
_exchange = _exchangeRegistry.getExchange(id);
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueRecoverer.java b/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueRecoverer.java
index fd836fd..621ea02 100644
--- a/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueRecoverer.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueRecoverer.java
@@ -104,13 +104,6 @@
public AMQQueue resolve()
{
String queueName = (String) _attributes.get(Queue.NAME);
- String owner = (String) _attributes.get(Queue.OWNER);
- boolean exclusive = (Boolean) _attributes.get(Queue.EXCLUSIVE);
-
- Map<String, Object> queueArgumentsMap = new LinkedHashMap<String, Object>(_attributes);
- queueArgumentsMap.remove(Queue.NAME);
- queueArgumentsMap.remove(Queue.OWNER);
- queueArgumentsMap.remove(Queue.EXCLUSIVE);
try
{
@@ -122,8 +115,9 @@
if (_queue == null)
{
- _queue = _queueFactory.restoreQueue(_id, queueName, owner, false, exclusive,
- false, queueArgumentsMap);
+ Map<String, Object> attributes = new LinkedHashMap<String, Object>(_attributes);
+ attributes.put(Queue.ID, _id);
+ _queue = _queueFactory.restoreQueue(attributes);
}
}
catch (QpidSecurityException e)
diff --git a/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java b/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
index 7cd9045..9996684 100755
--- a/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
+++ b/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
@@ -33,6 +33,7 @@
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.plugin.ExchangeType;
+import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.LinkRegistry;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.security.QpidSecurityException;
@@ -59,14 +60,7 @@
int removeQueue(AMQQueue queue) throws QpidSecurityException;
- AMQQueue createQueue(UUID id,
- String queueName,
- boolean durable,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- boolean deleteOnNoConsumer,
- Map<String, Object> arguments) throws QueueExistsException, QpidSecurityException;
+ AMQQueue createQueue(final AMQSessionModel creatingSession, Map<String, Object> arguments) throws QueueExistsException, QpidSecurityException;
Exchange createExchange(UUID id,
@@ -130,4 +124,20 @@
public void block();
public void unblock();
+
+ long getDefaultAlertThresholdMessageAge();
+
+ long getDefaultAlertThresholdMessageSize();
+
+ long getDefaultAlertThresholdQueueDepthMessages();
+
+ long getDefaultAlertThresholdQueueDepthBytes();
+
+ long getDefaultAlertRepeatGap();
+
+ long getDefaultQueueFlowControlSizeBytes();
+
+ long getDefaultQueueFlowResumeSizeBytes();
+
+ int getDefaultMaximumDeliveryAttempts();
}
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/configuration/QueueConfigurationTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/configuration/QueueConfigurationTest.java
index 96ac92a..2c3420a 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/configuration/QueueConfigurationTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/configuration/QueueConfigurationTest.java
@@ -137,9 +137,6 @@
qConf = new QueueConfiguration("test", vhostConfig);
assertEquals(2, qConf.getMaximumMessageAge());
- // Check inherited value
- qConf = new QueueConfiguration("test", _fullHostConf);
- assertEquals(1, qConf.getMaximumMessageAge());
}
public void testGetMaximumQueueDepth() throws ConfigurationException
@@ -153,9 +150,6 @@
qConf = new QueueConfiguration("test", vhostConfig);
assertEquals(2, qConf.getMaximumQueueDepth());
- // Check inherited value
- qConf = new QueueConfiguration("test", _fullHostConf);
- assertEquals(1, qConf.getMaximumQueueDepth());
}
public void testGetMaximumMessageSize() throws ConfigurationException
@@ -169,9 +163,6 @@
qConf = new QueueConfiguration("test", vhostConfig);
assertEquals(2, qConf.getMaximumMessageSize());
- // Check inherited value
- qConf = new QueueConfiguration("test", _fullHostConf);
- assertEquals(1, qConf.getMaximumMessageSize());
}
public void testGetMaximumMessageCount() throws ConfigurationException
@@ -185,22 +176,11 @@
qConf = new QueueConfiguration("test", vhostConfig);
assertEquals(2, qConf.getMaximumMessageCount());
- // Check inherited value
- qConf = new QueueConfiguration("test", _fullHostConf);
- assertEquals(1, qConf.getMaximumMessageCount());
}
public void testGetMinimumAlertRepeatGap() throws Exception
{
- // set broker attribute ALERT_REPEAT_GAP to 10
- when(_broker.getAttribute(Broker.QUEUE_ALERT_REPEAT_GAP)).thenReturn(10);
-
- // check that broker level setting is available on queue configuration
QueueConfiguration qConf = new QueueConfiguration("test", _emptyConf);
- assertEquals(10, qConf.getMinimumAlertRepeatGap());
-
- // remove configuration for ALERT_REPEAT_GAP on broker level
- when(_broker.getAttribute(Broker.QUEUE_ALERT_REPEAT_GAP)).thenReturn(null);
// Check default value
qConf = new QueueConfiguration("test", _emptyConf);
@@ -211,9 +191,6 @@
qConf = new QueueConfiguration("test", vhostConfig);
assertEquals(2, qConf.getMinimumAlertRepeatGap());
- // Check inherited value
- qConf = new QueueConfiguration("test", _fullHostConf);
- assertEquals(1, qConf.getMinimumAlertRepeatGap());
}
public void testSortQueueConfiguration() throws ConfigurationException
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java b/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
index 5a12a41..1358798 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
@@ -35,8 +35,10 @@
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.stats.StatisticsCounter;
+import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.StateChangeListener;
+import java.security.Principal;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
@@ -333,14 +335,26 @@
}
@Override
- public int compareTo(AMQSessionModel o)
+ public void close(AMQConstant cause, String message)
{
- return getId().compareTo(o.getId());
}
@Override
- public void close(AMQConstant cause, String message)
+ public void addDeleteTask(final Action task)
{
+
+ }
+
+ @Override
+ public void removeDeleteTask(final Action task)
+ {
+
+ }
+
+ @Override
+ public int compareTo(final Object o)
+ {
+ return 0;
}
}
@@ -431,12 +445,6 @@
}
@Override
- public String getUserName()
- {
- return null;
- }
-
- @Override
public boolean isSessionNameUnique(byte[] name)
{
return false;
@@ -455,6 +463,12 @@
}
@Override
+ public String getRemoteContainerName()
+ {
+ return null;
+ }
+
+ @Override
public String getClientVersion()
{
return null;
@@ -467,7 +481,7 @@
}
@Override
- public String getPrincipalAsString()
+ public Principal getAuthorizedPrincipal()
{
return null;
}
@@ -512,5 +526,17 @@
{
return null;
}
+
+ @Override
+ public void addDeleteTask(final Action task)
+ {
+
+ }
+
+ @Override
+ public void removeDeleteTask(final Action task)
+ {
+
+ }
}
}
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java
index 573f4a4..e8a2223 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java
@@ -20,17 +20,23 @@
*/
package org.apache.qpid.server.exchange;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+
import junit.framework.Assert;
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.server.util.BrokerTestHelper;
+import org.apache.qpid.server.virtualhost.QueueExistsException;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -70,10 +76,17 @@
}
}
+ private AMQQueue<?,?,?> createQueue(String name) throws QpidSecurityException, QueueExistsException
+ {
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUIDGenerator.generateRandomUUID());
+ attributes.put(Queue.NAME, name);
+ return _vhost.createQueue(null, attributes);
+ }
+
public void testNoRoute() throws Exception
{
- AMQQueue<?,?,?> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a*#b", false, null, false, false,
- false, null);
+ AMQQueue<?,?,?> queue = createQueue("a*#b");
_exchange.registerQueue(new Binding(null, "a.*.#.b",queue, _exchange, null));
@@ -84,8 +97,7 @@
public void testDirectMatch() throws Exception
{
- AMQQueue<?,?,?> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "ab", false, null, false, false,
- false, null);
+ AMQQueue<?,?,?> queue = createQueue("ab");
_exchange.registerQueue(new Binding(null, "a.b",queue, _exchange, null));
@@ -107,7 +119,7 @@
public void testStarMatch() throws Exception
{
- AMQQueue<?,?,?> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a*", false, null, false, false, false, null);
+ AMQQueue<?,?,?> queue = createQueue("a*");
_exchange.registerQueue(new Binding(null, "a.*",queue, _exchange, null));
@@ -138,7 +150,7 @@
public void testHashMatch() throws Exception
{
- AMQQueue<?,?,?> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false, false, null);
+ AMQQueue<?,?,?> queue = createQueue("a#");
_exchange.registerQueue(new Binding(null, "a.#",queue, _exchange, null));
@@ -189,8 +201,7 @@
public void testMidHash() throws Exception
{
- AMQQueue<?,?,?> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a", false, null, false, false,
- false, null);
+ AMQQueue<?,?,?> queue = createQueue("a");
_exchange.registerQueue(new Binding(null, "a.*.#.b",queue, _exchange, null));
routeMessage("a.c.d.b",0l);
@@ -215,8 +226,7 @@
public void testMatchAfterHash() throws Exception
{
- AMQQueue<?,?,?> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false,
- false, null);
+ AMQQueue<?,?,?> queue = createQueue("a#");
_exchange.registerQueue(new Binding(null, "a.*.#.b.c",queue, _exchange, null));
@@ -254,8 +264,7 @@
public void testHashAfterHash() throws Exception
{
- AMQQueue<?,?,?> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false,
- false, null);
+ AMQQueue<?,?,?> queue = createQueue("a#");
_exchange.registerQueue(new Binding(null, "a.*.#.b.c.#.d",queue, _exchange, null));
int queueCount = routeMessage("a.c.b.b.c",0l);
@@ -276,8 +285,7 @@
public void testHashHash() throws Exception
{
- AMQQueue<?,?,?> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false,
- false, null);
+ AMQQueue<?,?,?> queue = createQueue("a#");
_exchange.registerQueue(new Binding(null, "a.#.*.#.d",queue, _exchange, null));
int queueCount = routeMessage("a.c.b.b.c",0l);
@@ -298,8 +306,7 @@
public void testSubMatchFails() throws Exception
{
- AMQQueue queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a", false, null, false, false,
- false, null);
+ AMQQueue queue = createQueue("a");
_exchange.registerQueue(new Binding(null, "a.b.c.d",queue, _exchange, null));
int queueCount = routeMessage("a.b.c",0l);
@@ -328,8 +335,7 @@
public void testMoreRouting() throws Exception
{
- AMQQueue queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a", false, null, false, false,
- false, null);
+ AMQQueue queue = createQueue("a");
_exchange.registerQueue(new Binding(null, "a.b",queue, _exchange, null));
@@ -342,8 +348,7 @@
public void testMoreQueue() throws Exception
{
- AMQQueue queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a", false, null, false, false,
- false, null);
+ AMQQueue queue = createQueue("a");
_exchange.registerQueue(new Binding(null, "a.b",queue, _exchange, null));
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubjectTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubjectTest.java
index 13e6375..4d38025 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubjectTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubjectTest.java
@@ -22,6 +22,8 @@
import org.apache.qpid.server.protocol.AMQConnectionModel;
+import java.security.Principal;
+
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -43,9 +45,12 @@
{
super.setUp();
+ final Principal principal = mock(Principal.class);
+ when(principal.getName()).thenReturn(USER);
+
_connection = mock(AMQConnectionModel.class);
when(_connection.getConnectionId()).thenReturn(CONNECTION_ID);
- when(_connection.getPrincipalAsString()).thenReturn(USER);
+ when(_connection.getAuthorizedPrincipal()).thenReturn(principal);
when(_connection.getRemoteAddressString()).thenReturn("/"+IP_STRING);
when(_connection.getVirtualHostName()).thenReturn(VHOST);
_subject = new ConnectionLogSubject(_connection);
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/AMQQueueFactoryTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/AMQQueueFactoryTest.java
index 35ffd08..4b6a51f 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/AMQQueueFactoryTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/AMQQueueFactoryTest.java
@@ -20,17 +20,12 @@
*/
package org.apache.qpid.server.queue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyBoolean;
-import static org.mockito.Matchers.anyMap;
-import static org.mockito.Matchers.anyString;
-import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.*;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -45,9 +40,10 @@
import org.apache.qpid.server.logging.LogActor;
import org.apache.qpid.server.logging.RootMessageLogger;
import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.Queue;
-import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.plugin.ExchangeType;
+import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -98,33 +94,19 @@
private void delegateVhostQueueCreation() throws Exception
{
- final ArgumentCaptor<UUID> id = ArgumentCaptor.forClass(UUID.class);
- final ArgumentCaptor<String> queueName = ArgumentCaptor.forClass(String.class);
- final ArgumentCaptor<Boolean> durable = ArgumentCaptor.forClass(Boolean.class);
- final ArgumentCaptor<String> owner = ArgumentCaptor.forClass(String.class);
- final ArgumentCaptor<Boolean> autoDelete = ArgumentCaptor.forClass(Boolean.class);
- final ArgumentCaptor<Boolean> exclusive = ArgumentCaptor.forClass(Boolean.class);
- final ArgumentCaptor<Boolean> deleteOnNoConsumer = ArgumentCaptor.forClass(Boolean.class);
- final ArgumentCaptor<Map> arguments = ArgumentCaptor.forClass(Map.class);
- when(_virtualHost.createQueue(id.capture(), queueName.capture(), durable.capture(), owner.capture(),
- autoDelete.capture(), exclusive.capture(), deleteOnNoConsumer.capture(), arguments.capture())).then(
+ final ArgumentCaptor<Map> attributes = ArgumentCaptor.forClass(Map.class);
+
+ when(_virtualHost.createQueue(any(AMQSessionModel.class), attributes.capture())).then(
new Answer<AMQQueue>()
{
@Override
public AMQQueue answer(InvocationOnMock invocation) throws Throwable
{
- return _queueFactory.createQueue(id.getValue(),
- queueName.getValue(),
- durable.getValue(),
- owner.getValue(),
- autoDelete.getValue(),
- exclusive.getValue(),
- deleteOnNoConsumer.getValue(),
- arguments.getValue());
+ return _queueFactory.createQueue(null, attributes.getValue());
}
}
- );
+ );
}
private void mockQueueRegistry()
@@ -217,17 +199,14 @@
public void testPriorityQueueRegistration() throws Exception
{
- Map<String,Object> attributes = Collections.singletonMap(Queue.PRIORITIES, (Object) 5);
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, "testPriorityQueue");
+
+ attributes.put(Queue.PRIORITIES, 5);
- AMQQueue queue = _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(),
- "testPriorityQueue",
- false,
- "owner",
- false,
- false,
- false,
- attributes);
+ AMQQueue queue = _queueFactory.createQueue(null, attributes);
assertEquals("Queue not a priority queue", PriorityQueue.class, queue.getClass());
verifyQueueRegistered("testPriorityQueue");
@@ -240,10 +219,12 @@
String queueName = getName();
String dlQueueName = queueName + AMQQueueFactory.DEFAULT_DLQ_NAME_SUFFIX;
- AMQQueue queue = _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(), queueName, false, "owner", false,
- false,
- false,
- null);
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, queueName);
+
+
+ AMQQueue queue = _queueFactory.createQueue(null, attributes);
assertEquals("Queue not a simple queue", StandardQueue.class, queue.getClass());
verifyQueueRegistered(queueName);
@@ -261,7 +242,6 @@
*/
public void testDeadLetterQueueEnabled() throws Exception
{
- Map<String,Object> attributes = Collections.singletonMap(Queue.CREATE_DLQ_ON_CREATION, (Object) true);
String queueName = "testDeadLetterQueueEnabled";
String dlExchangeName = queueName + DefaultExchangeFactory.DEFAULT_DLE_NAME_SUFFIX;
@@ -270,14 +250,13 @@
assertNull("The DLQ should not yet exist", _virtualHost.getQueue(dlQueueName));
assertNull("The alternate exchange should not yet exist", _virtualHost.getExchange(dlExchangeName));
- AMQQueue queue = _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(),
- queueName,
- false,
- "owner",
- false,
- false,
- false,
- attributes);
+ Map<String,Object> attributes = new HashMap<String, Object>();
+
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, queueName);
+ attributes.put(Queue.CREATE_DLQ_ON_CREATION, true);
+
+ AMQQueue queue = _queueFactory.createQueue(null, attributes);
Exchange altExchange = queue.getAlternateExchange();
assertNotNull("Queue should have an alternate exchange as DLQ is enabled", altExchange);
@@ -314,14 +293,11 @@
assertNull("The DLQ should not yet exist", _virtualHost.getQueue(dlQueueName));
assertNull("The alternate exchange should not yet exist", _virtualHost.getExchange(dlExchangeName));
- AMQQueue queue = _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(),
- queueName,
- false,
- "owner",
- false,
- false,
- false,
- null);
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, queueName);
+
+ AMQQueue queue = _queueFactory.createQueue(null, attributes);
assertEquals("Unexpected maximum delivery count", 5, queue.getMaximumDeliveryCount());
Exchange altExchange = queue.getAlternateExchange();
@@ -348,7 +324,8 @@
*/
public void testDeadLetterQueueDisabled() throws Exception
{
- Map<String,Object> attributes = Collections.singletonMap(Queue.CREATE_DLQ_ON_CREATION, (Object) false);
+ Map<String,Object> attributes = new HashMap<String, Object>();
+
String queueName = "testDeadLetterQueueDisabled";
String dlExchangeName = queueName + DefaultExchangeFactory.DEFAULT_DLE_NAME_SUFFIX;
@@ -357,14 +334,11 @@
assertNull("The DLQ should not yet exist", _virtualHost.getQueue(dlQueueName));
assertNull("The alternate exchange should not exist", _virtualHost.getExchange(dlExchangeName));
- AMQQueue queue = _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(),
- queueName,
- false,
- "owner",
- false,
- false,
- false,
- attributes);
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, queueName);
+ attributes.put(Queue.CREATE_DLQ_ON_CREATION, false);
+
+ AMQQueue queue = _queueFactory.createQueue(null, attributes);
assertNull("Queue should not have an alternate exchange as DLQ is disabled", queue.getAlternateExchange());
assertNull("The alternate exchange should still not exist", _virtualHost.getExchange(dlExchangeName));
@@ -382,7 +356,6 @@
*/
public void testDeadLetterQueueNotCreatedForAutodeleteQueues() throws Exception
{
- Map<String,Object> attributes = Collections.singletonMap(Queue.CREATE_DLQ_ON_CREATION, (Object) true);
String queueName = "testDeadLetterQueueNotCreatedForAutodeleteQueues";
String dlExchangeName = queueName + DefaultExchangeFactory.DEFAULT_DLE_NAME_SUFFIX;
@@ -391,16 +364,18 @@
assertNull("The DLQ should not yet exist", _virtualHost.getQueue(dlQueueName));
assertNull("The alternate exchange should not exist", _virtualHost.getExchange(dlExchangeName));
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, queueName);
+
+ attributes.put(Queue.CREATE_DLQ_ON_CREATION, true);
+ attributes.put(Queue.LIFETIME_POLICY, LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS);
+
//create an autodelete queue
- AMQQueue queue = _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(),
- queueName,
- false,
- "owner",
- true,
- false,
- false,
- attributes);
- assertTrue("Queue should be autodelete", queue.isAutoDelete());
+ AMQQueue queue = _queueFactory.createQueue(null, attributes);
+ assertEquals("Queue should be autodelete",
+ LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS,
+ queue.getLifetimePolicy());
//ensure that the autodelete property overrides the request to enable DLQ
assertNull("Queue should not have an alternate exchange as queue is autodelete", queue.getAlternateExchange());
@@ -417,16 +392,13 @@
*/
public void testMaximumDeliveryCount() throws Exception
{
- Map<String,Object> attributes = Collections.singletonMap(Queue.MAXIMUM_DELIVERY_ATTEMPTS, (Object) 5);
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, "testMaximumDeliveryCount");
- final AMQQueue queue = _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(),
- "testMaximumDeliveryCount",
- false,
- "owner",
- false,
- false,
- false,
- attributes);
+ attributes.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, (Object) 5);
+
+ final AMQQueue queue = _queueFactory.createQueue(null, attributes);
assertNotNull("The queue was not registered as expected ", queue);
assertEquals("Maximum delivery count not as expected", 5, queue.getMaximumDeliveryCount());
@@ -440,14 +412,11 @@
*/
public void testMaximumDeliveryCountDefault() throws Exception
{
- final AMQQueue queue = _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(),
- "testMaximumDeliveryCount",
- false,
- "owner",
- false,
- false,
- false,
- null);
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, "testMaximumDeliveryCountDefault");
+
+ final AMQQueue queue = _queueFactory.createQueue(null, attributes);
assertNotNull("The queue was not registered as expected ", queue);
assertEquals("Maximum delivery count not as expected", 0, queue.getMaximumDeliveryCount());
@@ -462,15 +431,16 @@
{
try
{
- _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(), null, false, "owner", true, false,
- false,
- null);
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+
+ _queueFactory.createQueue(null, attributes);
fail("queue with null name can not be created!");
}
catch (Exception e)
{
assertTrue(e instanceof IllegalArgumentException);
- assertEquals("Queue name must not be null", e.getMessage());
+ assertEquals("Value for attribute name is not found", e.getMessage());
}
}
@@ -486,9 +456,14 @@
// change DLQ name to make its length bigger than exchange name
setTestSystemProperty(BrokerProperties.PROPERTY_DEAD_LETTER_EXCHANGE_SUFFIX, "_DLE");
setTestSystemProperty(BrokerProperties.PROPERTY_DEAD_LETTER_QUEUE_SUFFIX, "_DLQUEUE");
- Map<String,Object> attributes = Collections.singletonMap(Queue.CREATE_DLQ_ON_CREATION, (Object) true);
- _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(), queueName, false, "owner",
- false, false, false, attributes);
+
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, queueName);
+
+ attributes.put(Queue.CREATE_DLQ_ON_CREATION, true);
+
+ _queueFactory.createQueue(null, attributes);
fail("queue with DLQ name having more than 255 characters can not be created!");
}
catch (Exception e)
@@ -511,9 +486,14 @@
// change DLQ name to make its length bigger than exchange name
setTestSystemProperty(BrokerProperties.PROPERTY_DEAD_LETTER_EXCHANGE_SUFFIX, "_DLEXCHANGE");
setTestSystemProperty(BrokerProperties.PROPERTY_DEAD_LETTER_QUEUE_SUFFIX, "_DLQ");
- Map<String,Object> attributes = Collections.singletonMap(Queue.CREATE_DLQ_ON_CREATION, (Object) true);
- _queueFactory.createQueue(UUIDGenerator.generateRandomUUID(), queueName, false, "owner",
- false, false, false, attributes);
+
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, queueName);
+
+ attributes.put(Queue.CREATE_DLQ_ON_CREATION, (Object) true);
+
+ _queueFactory.createQueue(null, attributes);
fail("queue with DLE name having more than 255 characters can not be created!");
}
catch (Exception e)
@@ -528,6 +508,7 @@
{
Map<String,String> arguments = new HashMap<String, String>();
+
arguments.put(QueueArgumentsConverter.QPID_GROUP_HEADER_KEY,"mykey");
arguments.put(QueueArgumentsConverter.QPID_SHARED_MSG_GROUP,"1");
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/ConflationQueueListTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/ConflationQueueListTest.java
index c2291f5..42b83d8 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/ConflationQueueListTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/ConflationQueueListTest.java
@@ -26,9 +26,11 @@
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.virtualhost.VirtualHost;
-import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import java.util.UUID;
public class ConflationQueueListTest extends TestCase
@@ -46,8 +48,11 @@
protected void setUp() throws Exception
{
super.setUp();
- _queue = new ConflationQueue(UUID.randomUUID(), getName(), false, null, false, false, mock(VirtualHost.class),
- Collections.<String,Object>emptyMap(),CONFLATION_KEY);
+ Map<String,Object> queueAttributes = new HashMap<String, Object>();
+ queueAttributes.put(Queue.ID, UUID.randomUUID());
+ queueAttributes.put(Queue.NAME, getName());
+ queueAttributes.put(Queue.LVQ_KEY, CONFLATION_KEY);
+ _queue = new ConflationQueue(mock(VirtualHost.class), null, queueAttributes);
_list = _queue.getEntries();
}
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
index c462468..91c0136 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
@@ -27,6 +27,7 @@
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.security.AuthorizationHolder;
import org.apache.qpid.server.consumer.Consumer;
@@ -131,6 +132,12 @@
return 0;
}
+ @Override
+ public LifetimePolicy getLifetimePolicy()
+ {
+ return null;
+ }
+
public int getBindingCountHigh()
{
return 0;
@@ -216,7 +223,7 @@
}
@Override
- public void addQueueDeleteTask(final Action task)
+ public void addDeleteTask(final Action task)
{
}
@@ -345,7 +352,7 @@
}
@Override
- public void removeQueueDeleteTask(final Action task)
+ public void removeDeleteTask(final Action task)
{
}
@@ -478,6 +485,12 @@
return false;
}
+ @Override
+ public boolean verifySessionAccess(final AMQSessionModel session)
+ {
+ return false;
+ }
+
public Exchange getAlternateExchange()
{
return null;
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/PriorityQueueListTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/PriorityQueueListTest.java
index 3db5d0f..0ac4eb8 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/PriorityQueueListTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/PriorityQueueListTest.java
@@ -26,10 +26,12 @@
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
-import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import java.util.UUID;
public class PriorityQueueListTest extends QpidTestCase
@@ -45,16 +47,11 @@
protected void setUp()
{
QueueEntry[] entries = new QueueEntry[PRIORITIES.length];
-
- PriorityQueue queue = new PriorityQueue(UUID.randomUUID(),
- getName(),
- false,
- null,
- false,
- false,
- mock(VirtualHost.class),
- Collections.<String,Object>emptyMap(),
- 10);
+ Map<String,Object> queueAttributes = new HashMap<String, Object>();
+ queueAttributes.put(Queue.ID, UUID.randomUUID());
+ queueAttributes.put(Queue.NAME, getName());
+ queueAttributes.put(Queue.PRIORITIES, 10);
+ PriorityQueue queue = new PriorityQueue(mock(VirtualHost.class), null, queueAttributes);
_list = queue.getEntries();
for (int i = 0; i < PRIORITIES.length; i++)
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java
index 66e4286..139e324 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java
@@ -27,11 +27,13 @@
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.message.MessageInstance.EntryState;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.lang.reflect.Field;
-import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import java.util.UUID;
import static org.mockito.Mockito.mock;
@@ -199,8 +201,10 @@
{
int numberOfEntries = 5;
QueueEntryImpl[] entries = new QueueEntryImpl[numberOfEntries];
- StandardQueue queue = new StandardQueue(UUID.randomUUID(), getName(), false, null, false, false,
- mock(VirtualHost.class), Collections.<String,Object>emptyMap());
+ Map<String,Object> queueAttributes = new HashMap<String, Object>();
+ queueAttributes.put(Queue.ID, UUID.randomUUID());
+ queueAttributes.put(Queue.NAME, getName());
+ StandardQueue queue = new StandardQueue(mock(VirtualHost.class), null, queueAttributes);
OrderedQueueEntryList queueEntryList = queue.getEntries();
// create test entries
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTestBase.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTestBase.java
index 1934349..956a9fc 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTestBase.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTestBase.java
@@ -29,12 +29,12 @@
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.when;
-import java.util.Arrays;
-import java.util.EnumSet;
-import java.util.Map;
+import java.util.*;
import org.apache.log4j.Logger;
import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.server.exchange.DirectExchange;
@@ -51,10 +51,6 @@
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
abstract class SimpleAMQQueueTestBase<E extends QueueEntryImpl<E,Q,L>, Q extends SimpleAMQQueue<E,Q,L>, L extends SimpleQueueEntryList<E,Q,L>> extends QpidTestCase
{
private static final Logger _logger = Logger.getLogger(SimpleAMQQueueTestBase.class);
@@ -68,7 +64,7 @@
private DirectExchange _exchange;
private MockConsumer _consumerTarget = new MockConsumer();
private QueueConsumer _consumer;
- private Map<String,Object> _arguments = null;
+ private Map<String,Object> _arguments = Collections.emptyMap();
@Override
public void setUp() throws Exception
@@ -78,8 +74,12 @@
_virtualHost = BrokerTestHelper.createVirtualHost(getClass().getName());
- _queue = (Q) _virtualHost.createQueue(UUIDGenerator.generateRandomUUID(), _qname, false, _owner,
- false, false, false, _arguments);
+ Map<String,Object> attributes = new HashMap<String, Object>(_arguments);
+ attributes.put(Queue.ID, UUIDGenerator.generateRandomUUID());
+ attributes.put(Queue.NAME, _qname);
+ attributes.put(Queue.OWNER, _owner);
+
+ _queue = (Q) _virtualHost.createQueue(null, attributes);
_exchange = (DirectExchange) _virtualHost.getExchange(ExchangeDefaults.DIRECT_EXCHANGE_NAME);
}
@@ -104,9 +104,10 @@
_queue.stop();
try
{
- _queue = (Q) _virtualHost.createQueue(UUIDGenerator.generateRandomUUID(), null,
- false, _owner, false,
- false, false, _arguments);
+ Map<String,Object> attributes = new HashMap<String, Object>(_arguments);
+ attributes.put(Queue.ID, UUIDGenerator.generateRandomUUID());
+
+ _queue = (Q) _virtualHost.createQueue(null, attributes);
assertNull("Queue was created", _queue);
}
catch (IllegalArgumentException e)
@@ -115,10 +116,10 @@
e.getMessage().contains("name"));
}
- _queue = (Q) _virtualHost.createQueue(UUIDGenerator.generateRandomUUID(),
- "differentName", false,
- _owner, false,
- false, false, _arguments);
+ Map<String,Object> attributes = new HashMap<String, Object>(_arguments);
+ attributes.put(Queue.ID, UUIDGenerator.generateRandomUUID());
+ attributes.put(Queue.NAME, "differentName");
+ _queue = (Q) _virtualHost.createQueue(null, attributes);
assertNotNull("Queue was not created", _queue);
}
@@ -1137,15 +1138,17 @@
{
public NonAsyncDeliverQueue(final TestSimpleQueueEntryListFactory factory, VirtualHost vhost)
{
- super(UUIDGenerator.generateRandomUUID(),
- "testQueue",
- false,
- "testOwner",
- false,
- false,
- vhost,
- factory,
- null);
+ super(vhost, null, attributes(), factory);
+ }
+
+ private static Map<String,Object> attributes()
+ {
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, "test");
+ attributes.put(Queue.DURABLE, false);
+ attributes.put(Queue.LIFETIME_POLICY, LifetimePolicy.PERMANENT);
+ return attributes;
}
@Override
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueThreadPoolTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueThreadPoolTest.java
index c115af5..286c47f 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueThreadPoolTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueThreadPoolTest.java
@@ -21,11 +21,15 @@
package org.apache.qpid.server.queue;
import org.apache.qpid.pool.ReferenceCountingExecutorService;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.util.BrokerTestHelper;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
+import java.util.HashMap;
+import java.util.Map;
+
public class SimpleAMQQueueThreadPoolTest extends QpidTestCase
{
@@ -50,8 +54,10 @@
try
{
- SimpleAMQQueue queue = (SimpleAMQQueue)
- test.createQueue(UUIDGenerator.generateRandomUUID(), "test", false, "owner", false, false, false, null);
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUIDGenerator.generateRandomUUID());
+ attributes.put(Queue.NAME, "test");
+ AMQQueue queue = test.createQueue(null, attributes);
assertFalse("Creation did not start Pool.", ReferenceCountingExecutorService.getInstance().getPool().isShutdown());
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleQueueEntryImplTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleQueueEntryImplTest.java
index 9457d59..7aa546e 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleQueueEntryImplTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleQueueEntryImplTest.java
@@ -22,8 +22,11 @@
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.virtualhost.VirtualHost;
+import java.util.HashMap;
+import java.util.Map;
import java.util.UUID;
import static org.mockito.Mockito.mock;
@@ -38,8 +41,10 @@
public void setUp() throws Exception
{
mockLogging();
-
- StandardQueue queue = new StandardQueue(UUID.randomUUID(), "SimpleQueueEntryImplTest", false, null,false, false, mock(VirtualHost.class),null);
+ Map<String,Object> queueAttributes = new HashMap<String, Object>();
+ queueAttributes.put(Queue.ID, UUID.randomUUID());
+ queueAttributes.put(Queue.NAME, "SimpleQueueEntryImplTest");
+ StandardQueue queue = new StandardQueue(mock(VirtualHost.class), null, queueAttributes);
queueEntryList = queue.getEntries();
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryListTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryListTest.java
index f8953bb..cfa58f1 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryListTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryListTest.java
@@ -26,9 +26,13 @@
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
import java.util.UUID;
import static org.mockito.Matchers.eq;
@@ -74,8 +78,15 @@
{
mockLogging();
+ Map<String,Object> attributes = new HashMap<String,Object>();
+ attributes.put(Queue.ID,UUID.randomUUID());
+ attributes.put(Queue.NAME, getName());
+ attributes.put(Queue.DURABLE, false);
+ attributes.put(Queue.LIFETIME_POLICY, LifetimePolicy.PERMANENT);
+ attributes.put(Queue.SORT_KEY, "KEY");
+
// Create test list
- _testQueue = new SortedQueue(UUID.randomUUID(), getName(), false, null, false,false, mock(VirtualHost.class), null, "KEY", new QueueEntryListFactory<SortedQueueEntry,SortedQueue,SortedQueueEntryList>()
+ _testQueue = new SortedQueue(mock(VirtualHost.class), null, attributes, new QueueEntryListFactory<SortedQueueEntry,SortedQueue,SortedQueueEntryList>()
{
@Override
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryTest.java
index 25d6dbb..e558751 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryTest.java
@@ -20,11 +20,15 @@
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.message.AMQMessageHeader;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.virtualhost.VirtualHost;
import static org.mockito.Matchers.eq;
@@ -42,7 +46,15 @@
public void setUp() throws Exception
{
mockLogging();
- SortedQueue queue = new SortedQueue(UUID.randomUUID(), getName(), false, null, false,false, mock(VirtualHost.class), null, "KEY", new QueueEntryListFactory<SortedQueueEntry,SortedQueue,SortedQueueEntryList>()
+
+ Map<String,Object> attributes = new HashMap<String,Object>();
+ attributes.put(Queue.ID,UUID.randomUUID());
+ attributes.put(Queue.NAME, getName());
+ attributes.put(Queue.DURABLE, false);
+ attributes.put(Queue.LIFETIME_POLICY, LifetimePolicy.PERMANENT);
+ attributes.put(Queue.SORT_KEY, "KEY");
+
+ SortedQueue queue = new SortedQueue(mock(VirtualHost.class), null, attributes, new QueueEntryListFactory<SortedQueueEntry,SortedQueue,SortedQueueEntryList>()
{
@Override
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/StandardQueueEntryListTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/StandardQueueEntryListTest.java
index 2c697cf..fe9273d 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/StandardQueueEntryListTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/StandardQueueEntryListTest.java
@@ -22,9 +22,10 @@
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.virtualhost.VirtualHost;
-import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -45,8 +46,11 @@
protected void setUp()
{
oldScavengeValue = System.setProperty(SCAVENGE_PROP, "9");
- _testQueue = new StandardQueue(UUID.randomUUID(),getName(),false,null,false,false,mock(VirtualHost.class),
- Collections.<String,Object>emptyMap());
+
+ Map<String,Object> queueAttributes = new HashMap<String, Object>();
+ queueAttributes.put(Queue.ID, UUID.randomUUID());
+ queueAttributes.put(Queue.NAME, getName());
+ _testQueue = new StandardQueue(mock(VirtualHost.class), null, queueAttributes);
_sqel = _testQueue.getEntries();
for(int i = 1; i <= 100; i++)
@@ -86,9 +90,10 @@
{
if(newList)
{
- StandardQueue queue =
- new StandardQueue(UUID.randomUUID(), getName(), false, null, false, false, mock(VirtualHost.class),
- Collections.<String, Object>emptyMap());
+ Map<String,Object> queueAttributes = new HashMap<String, Object>();
+ queueAttributes.put(Queue.ID, UUID.randomUUID());
+ queueAttributes.put(Queue.NAME, getName());
+ StandardQueue queue = new StandardQueue(mock(VirtualHost.class), null, queueAttributes);
return queue.getEntries();
}
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/queue/StandardQueueTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/queue/StandardQueueTest.java
index c7b812e..77886fa 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/queue/StandardQueueTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/queue/StandardQueueTest.java
@@ -25,18 +25,21 @@
import org.apache.qpid.server.consumer.MockConsumer;
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.ServerMessage;
-import org.apache.qpid.server.model.UUIDGenerator;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Arrays;
-import java.util.Collections;
import java.util.EnumSet;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import static org.mockito.Mockito.mock;
+
public class StandardQueueTest extends SimpleAMQQueueTestBase<StandardQueueEntry, StandardQueue, StandardQueueEntryList>
{
@@ -44,7 +47,12 @@
{
try
{
- setQueue(new StandardQueue(UUIDGenerator.generateRandomUUID(), getQname(), false, getOwner(), false,false, null, getArguments()));
+ Map<String,Object> queueAttributes = new HashMap<String, Object>();
+ queueAttributes.put(Queue.ID, UUID.randomUUID());
+ queueAttributes.put(Queue.NAME, "testActiveConsumerCount");
+ queueAttributes.put(Queue.OWNER, "testOwner");
+
+ setQueue(new StandardQueue(null, null, queueAttributes));
assertNull("Queue was created", getQueue());
}
catch (IllegalArgumentException e)
@@ -58,7 +66,13 @@
public void testAutoDeleteQueue() throws Exception
{
getQueue().stop();
- setQueue(new StandardQueue(UUIDGenerator.generateRandomUUID(), getQname(), false, null, true, false, getVirtualHost(), Collections.<String,Object>emptyMap()));
+ Map<String,Object> queueAttributes = new HashMap<String, Object>();
+ queueAttributes.put(Queue.ID, UUID.randomUUID());
+ queueAttributes.put(Queue.NAME, getQname());
+ queueAttributes.put(Queue.LIFETIME_POLICY, LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS);
+ final StandardQueue queue = new StandardQueue(getVirtualHost(), null, queueAttributes);
+
+ setQueue(queue);
getQueue().setDeleteOnNoConsumers(true);
ServerMessage message = createMessage(25l);
@@ -75,8 +89,12 @@
public void testActiveConsumerCount() throws Exception
{
- final StandardQueue queue = new StandardQueue(UUIDGenerator.generateRandomUUID(), "testActiveConsumerCount", false,
- "testOwner", false, false, getVirtualHost(), null);
+
+ Map<String,Object> queueAttributes = new HashMap<String, Object>();
+ queueAttributes.put(Queue.ID, UUID.randomUUID());
+ queueAttributes.put(Queue.NAME, "testActiveConsumerCount");
+ queueAttributes.put(Queue.OWNER, "testOwner");
+ final StandardQueue queue = new StandardQueue(getVirtualHost(), null, queueAttributes);
//verify adding an active consumer increases the count
final MockConsumer consumer1 = new MockConsumer();
@@ -145,8 +163,7 @@
public void testEnqueueDequeuedEntry() throws Exception
{
// create a queue where each even entry is considered a dequeued
- SimpleAMQQueue queue = new DequeuedQueue(UUIDGenerator.generateRandomUUID(), "test", false,
- "testOwner", false, false, getVirtualHost(), null);
+ SimpleAMQQueue queue = new DequeuedQueue(getVirtualHost());
// create a consumer
MockConsumer consumer = new MockConsumer();
@@ -180,9 +197,11 @@
int messageNumber = 4;
int dequeueMessageIndex = 1;
+ Map<String,Object> queueAttributes = new HashMap<String, Object>();
+ queueAttributes.put(Queue.ID, UUID.randomUUID());
+ queueAttributes.put(Queue.NAME, "test");
// create queue with overridden method deliverAsync
- StandardQueue testQueue = new StandardQueue(UUIDGenerator.generateRandomUUID(), "test",
- false, "testOwner", false, false, getVirtualHost(), null)
+ StandardQueue testQueue = new StandardQueue(getVirtualHost(), null, queueAttributes)
{
@Override
public void deliverAsync(QueueConsumer sub)
@@ -249,16 +268,19 @@
private static class DequeuedQueue extends SimpleAMQQueue<DequeuedQueueEntry, DequeuedQueue, DequeuedQueueEntryList>
{
- public DequeuedQueue(final UUID id,
- final String queueName,
- final boolean durable,
- final String owner,
- final boolean autoDelete,
- final boolean exclusive,
- final VirtualHost virtualHost,
- final Map<String, Object> arguments)
+ public DequeuedQueue(VirtualHost virtualHost)
{
- super(id, queueName, durable, owner, autoDelete, exclusive, virtualHost, new DequeuedQueueEntryListFactory(), arguments);
+ super(virtualHost, null, attributes(), new DequeuedQueueEntryListFactory());
+ }
+
+ private static Map<String,Object> attributes()
+ {
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUID.randomUUID());
+ attributes.put(Queue.NAME, "test");
+ attributes.put(Queue.DURABLE, false);
+ attributes.put(Queue.LIFETIME_POLICY, LifetimePolicy.PERMANENT);
+ return attributes;
}
}
private static class DequeuedQueueEntryListFactory implements QueueEntryListFactory<DequeuedQueueEntry, DequeuedQueue, DequeuedQueueEntryList>
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java b/java/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java
index 8b66e7d..c5c68ec 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java
@@ -32,6 +32,7 @@
import java.io.File;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.UUID;
@@ -40,6 +41,7 @@
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.message.EnqueueableMessage;
+import org.apache.qpid.server.model.ExclusivityPolicy;
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.UUIDGenerator;
@@ -143,7 +145,7 @@
verify(_recoveryHandler).configuredObject(eq(_exchangeId), eq(EXCHANGE),
eq(map( org.apache.qpid.server.model.Exchange.NAME, getName(),
org.apache.qpid.server.model.Exchange.TYPE, getName()+"Type",
- org.apache.qpid.server.model.Exchange.LIFETIME_POLICY, LifetimePolicy.AUTO_DELETE.toString())));
+ org.apache.qpid.server.model.Exchange.LIFETIME_POLICY, LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS.name())));
}
private Map<String,Object> map(Object... vals)
@@ -220,7 +222,7 @@
Map<String, Object> queueAttributes = new HashMap<String, Object>();
queueAttributes.put(Queue.NAME, getName());
queueAttributes.put(Queue.OWNER, getName()+"Owner");
- queueAttributes.put(Queue.EXCLUSIVE, Boolean.TRUE);
+ queueAttributes.put(Queue.EXCLUSIVE, ExclusivityPolicy.CONTAINER.name());
verify(_recoveryHandler).configuredObject(eq(_queueId), eq(QUEUE), eq(queueAttributes));
}
@@ -240,7 +242,7 @@
queueAttributes.put(Queue.NAME, getName());
queueAttributes.put(Queue.OWNER, getName()+"Owner");
- queueAttributes.put(Queue.EXCLUSIVE, Boolean.TRUE);
+ queueAttributes.put(Queue.EXCLUSIVE, ExclusivityPolicy.CONTAINER.name());
queueAttributes.putAll(attributes);
verify(_recoveryHandler).configuredObject(eq(_queueId), eq(QUEUE), eq(queueAttributes));
@@ -258,7 +260,7 @@
Map<String, Object> queueAttributes = new HashMap<String, Object>();
queueAttributes.put(Queue.NAME, getName());
queueAttributes.put(Queue.OWNER, getName()+"Owner");
- queueAttributes.put(Queue.EXCLUSIVE, Boolean.TRUE);
+ queueAttributes.put(Queue.EXCLUSIVE, ExclusivityPolicy.CONTAINER.name());
queueAttributes.put(Queue.ALTERNATE_EXCHANGE, alternateExchange.getId().toString());
verify(_recoveryHandler).configuredObject(eq(_queueId), eq(QUEUE), eq(queueAttributes));
@@ -292,8 +294,6 @@
Map<String,Object> queueAttributes = new HashMap<String, Object>();
queueAttributes.put(Queue.NAME, getName());
- queueAttributes.put(Queue.OWNER, getName()+"Owner");
- queueAttributes.put(Queue.EXCLUSIVE, Boolean.FALSE);
queueAttributes.putAll(attributes);
verify(_recoveryHandler).configuredObject(eq(_queueId), eq(QUEUE), eq(queueAttributes));
@@ -320,8 +320,6 @@
Map<String,Object> queueAttributes = new HashMap<String, Object>();
queueAttributes.put(Queue.NAME, getName());
- queueAttributes.put(Queue.OWNER, getName()+"Owner");
- queueAttributes.put(Queue.EXCLUSIVE, Boolean.FALSE);
queueAttributes.putAll(attributes);
queueAttributes.put(Queue.ALTERNATE_EXCHANGE, alternateExchange.getId().toString());
@@ -361,13 +359,19 @@
{
AMQQueue queue = mock(AMQQueue.class);
when(queue.getName()).thenReturn(queueName);
- when(queue.getOwner()).thenReturn(queueOwner);
when(queue.isExclusive()).thenReturn(exclusive);
when(queue.getId()).thenReturn(_queueId);
when(queue.getAlternateExchange()).thenReturn(alternateExchange);
- if(arguments != null && !arguments.isEmpty())
+ final Map<String,Object> attributes = arguments == null ? new LinkedHashMap<String, Object>() : new LinkedHashMap<String, Object>(arguments);
+ attributes.put(Queue.NAME, queueName);
+ if(exclusive)
{
- when(queue.getAvailableAttributes()).thenReturn(arguments.keySet());
+ when(queue.getOwner()).thenReturn(queueOwner);
+
+ attributes.put(Queue.OWNER, queueOwner);
+ attributes.put(Queue.EXCLUSIVE, ExclusivityPolicy.CONTAINER);
+ }
+ when(queue.getAvailableAttributes()).thenReturn(attributes.keySet());
final ArgumentCaptor<String> requestedAttribute = ArgumentCaptor.forClass(String.class);
when(queue.getAttribute(requestedAttribute.capture())).then(
new Answer()
@@ -377,10 +381,9 @@
public Object answer(final InvocationOnMock invocation) throws Throwable
{
String attrName = requestedAttribute.getValue();
- return arguments.get(attrName);
+ return attributes.get(attrName);
}
});
- }
return queue;
}
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/util/BrokerTestHelper.java b/java/broker-core/src/test/java/org/apache/qpid/server/util/BrokerTestHelper.java
index fb63fef..f082d58 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/util/BrokerTestHelper.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/util/BrokerTestHelper.java
@@ -25,11 +25,13 @@
import static org.mockito.Mockito.when;
import java.net.SocketAddress;
-import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import java.util.UUID;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.configuration.VirtualHostConfiguration;
@@ -182,8 +184,10 @@
public static AMQQueue createQueue(String queueName, VirtualHost virtualHost)
throws QpidSecurityException, QueueExistsException
{
- AMQQueue queue = virtualHost.createQueue(UUIDGenerator.generateRandomUUID(), queueName, false, null,
- false, false, false, Collections.<String, Object>emptyMap());
+ Map<String,Object> attributes = new HashMap<String, Object>();
+ attributes.put(Queue.ID, UUIDGenerator.generateRandomUUID());
+ attributes.put(Queue.NAME, queueName);
+ AMQQueue queue = virtualHost.createQueue(null, attributes);
return queue;
}
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/DurableConfigurationRecovererTest.java b/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/DurableConfigurationRecovererTest.java
index 223e2c5..1e25aac 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/DurableConfigurationRecovererTest.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/DurableConfigurationRecovererTest.java
@@ -46,6 +46,7 @@
import org.apache.qpid.server.store.DurableConfigurationRecoverer;
import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.store.DurableConfiguredObjectRecoverer;
+import org.apache.qpid.server.util.MapValueConverter;
import org.apache.qpid.test.utils.QpidTestCase;
import org.mockito.ArgumentCaptor;
import org.mockito.invocation.InvocationOnMock;
@@ -117,14 +118,11 @@
- final ArgumentCaptor<UUID> idArg = ArgumentCaptor.forClass(UUID.class);
- final ArgumentCaptor<String> queueArg = ArgumentCaptor.forClass(String.class);
- final ArgumentCaptor<Map> argsArg = ArgumentCaptor.forClass(Map.class);
+ final ArgumentCaptor<Map> attributesArg = ArgumentCaptor.forClass(Map.class);
_queueFactory = mock(QueueFactory.class);
- when(_queueFactory.restoreQueue(idArg.capture(), queueArg.capture(),
- anyString(), anyBoolean(), anyBoolean(), anyBoolean(), argsArg.capture())).then(
+ when(_queueFactory.restoreQueue(attributesArg.capture())).then(
new Answer()
{
@@ -133,8 +131,9 @@
{
final AMQQueue queue = mock(AMQQueue.class);
- final String queueName = queueArg.getValue();
- final UUID queueId = idArg.getValue();
+ final Map attributes = attributesArg.getValue();
+ final String queueName = (String) attributes.get(Queue.NAME);
+ final UUID queueId = MapValueConverter.getUUIDAttribute(Queue.ID, attributes);
when(queue.getName()).thenReturn(queueName);
when(queue.getId()).thenReturn(queueId);
@@ -153,10 +152,10 @@
return null;
}
}
- ).when(queue).setAlternateExchange(altExchangeArg.capture());
+ ).when(queue).setAlternateExchange(altExchangeArg.capture());
- Map args = argsArg.getValue();
- if(args.containsKey(Queue.ALTERNATE_EXCHANGE))
+ Map args = attributes;
+ if (args.containsKey(Queue.ALTERNATE_EXCHANGE))
{
final UUID exchangeId = UUID.fromString(args.get(Queue.ALTERNATE_EXCHANGE).toString());
final Exchange exchange = _exchangeRegistry.getExchange(exchangeId);
@@ -470,7 +469,6 @@
{
queue.put(Queue.ALTERNATE_EXCHANGE, alternateExchangeId.toString());
}
- queue.put(Queue.EXCLUSIVE, false);
return queue;
diff --git a/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java b/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java
index f7eecc7..cee7edf 100644
--- a/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java
+++ b/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java
@@ -29,6 +29,7 @@
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.plugin.ExchangeType;
+import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.LinkRegistry;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
@@ -152,14 +153,7 @@
}
@Override
- public AMQQueue createQueue(UUID id,
- String queueName,
- boolean durable,
- String owner,
- boolean autoDelete,
- boolean exclusive,
- boolean deleteOnNoConsumer,
- Map<String, Object> arguments)
+ public AMQQueue createQueue(final AMQSessionModel creatingSession, Map<String, Object> arguments)
{
return null;
}
@@ -314,4 +308,52 @@
public void unblock()
{
}
+
+ @Override
+ public long getDefaultAlertThresholdMessageAge()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getDefaultAlertThresholdMessageSize()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getDefaultAlertThresholdQueueDepthMessages()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getDefaultAlertThresholdQueueDepthBytes()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getDefaultAlertRepeatGap()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getDefaultQueueFlowControlSizeBytes()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getDefaultQueueFlowResumeSizeBytes()
+ {
+ return 0;
+ }
+
+ @Override
+ public int getDefaultMaximumDeliveryAttempts()
+ {
+ return 0;
+ }
}
diff --git a/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java b/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
index ff7ce0a..9c012eb 100644
--- a/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
+++ b/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
@@ -25,6 +25,7 @@
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import javax.security.auth.Subject;
@@ -42,6 +43,7 @@
import org.apache.qpid.server.security.AuthorizationHolder;
import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
import org.apache.qpid.server.stats.StatisticsCounter;
+import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.transport.Connection;
@@ -56,7 +58,8 @@
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SOCKET_FORMAT;
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.USER_FORMAT;
-public class ServerConnection extends Connection implements AMQConnectionModel, LogSubject, AuthorizationHolder
+public class ServerConnection extends Connection implements AMQConnectionModel<ServerConnection, ServerSession>,
+ LogSubject, AuthorizationHolder
{
private Runnable _onOpenTask;
private AtomicBoolean _logClosed = new AtomicBoolean(false);
@@ -72,6 +75,10 @@
private AtomicLong _lastIoTime = new AtomicLong();
private boolean _blocking;
private Transport _transport;
+
+ private final CopyOnWriteArrayList<Action<? super ServerConnection>> _taskList =
+ new CopyOnWriteArrayList<Action<? super ServerConnection>>();
+
private volatile boolean _stopped;
public ServerConnection(final long connectionId, Broker broker)
@@ -197,7 +204,7 @@
_onOpenTask = task;
}
- public void closeSession(AMQSessionModel session, AMQConstant cause, String message)
+ public void closeSession(ServerSession session, AMQConstant cause, String message)
{
ExecutionException ex = new ExecutionException();
ExecutionErrorCode code = ExecutionErrorCode.INTERNAL_ERROR;
@@ -211,7 +218,7 @@
}
ex.setErrorCode(code);
ex.setDescription(message);
- ((ServerSession)session).invoke(ex);
+ session.invoke(ex);
session.close(cause, message);
}
@@ -315,6 +322,7 @@
public void close(AMQConstant cause, String message)
{
closeSubscriptions();
+ performDeleteTasks();
ConnectionCloseCode replyCode = ConnectionCloseCode.NORMAL;
try
{
@@ -327,6 +335,14 @@
close(replyCode, message);
}
+ protected void performDeleteTasks()
+ {
+ for(Action<? super ServerConnection> task : _taskList)
+ {
+ task.performAction(this);
+ }
+ }
+
public synchronized void block()
{
if(!_blocking)
@@ -367,12 +383,12 @@
super.removeSession(ssn);
}
- public List<AMQSessionModel> getSessionModels()
+ public List<ServerSession> getSessionModels()
{
- List<AMQSessionModel> sessions = new ArrayList<AMQSessionModel>();
+ List<ServerSession> sessions = new ArrayList<ServerSession>();
for (Session ssn : getChannels())
{
- sessions.add((AMQSessionModel) ssn);
+ sessions.add((ServerSession) ssn);
}
return sessions;
}
@@ -475,14 +491,10 @@
return String.valueOf(getRemoteAddress());
}
- public String getUserName()
- {
- return _authorizedPrincipal.getName();
- }
-
@Override
public void closed()
{
+ performDeleteTasks();
closeSubscriptions();
super.closed();
}
@@ -522,6 +534,12 @@
}
@Override
+ public String getRemoteContainerName()
+ {
+ return getConnectionDelegate().getClientId();
+ }
+
+ @Override
public String getClientVersion()
{
return getConnectionDelegate().getClientVersion();
@@ -533,11 +551,6 @@
return getConnectionDelegate().getClientProduct();
}
- public String getPrincipalAsString()
- {
- return getAuthorizedPrincipal() == null ? null : getAuthorizedPrincipal().getName();
- }
-
public long getSessionCountLimit()
{
return getChannelMax();
@@ -565,4 +578,16 @@
super.doHeartBeat();
}
+
+ @Override
+ public void addDeleteTask(final Action<? super ServerConnection> task)
+ {
+ _taskList.add(task);
+ }
+
+ @Override
+ public void removeDeleteTask(final Action<? super ServerConnection> task)
+ {
+ _taskList.remove(task);
+ }
}
diff --git a/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java b/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
index c85a415..dc26249 100644
--- a/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
+++ b/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.server.protocol.v0_10;
+import java.security.Principal;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -310,14 +311,18 @@
private boolean isSessionNameUnique(final byte[] name, final Connection conn)
{
final ServerConnection sconn = (ServerConnection) conn;
- final String userId = sconn.getUserName();
+ final Principal authorizedPrincipal = sconn.getAuthorizedPrincipal();
+ final String userId = authorizedPrincipal == null ? "" : authorizedPrincipal.getName();
final Iterator<AMQConnectionModel> connections =
((ServerConnection)conn).getVirtualHost().getConnectionRegistry().getConnections().iterator();
while(connections.hasNext())
{
- final AMQConnectionModel amqConnectionModel = (AMQConnectionModel) connections.next();
- if (userId.equals(amqConnectionModel.getUserName()) && !amqConnectionModel.isSessionNameUnique(name))
+ final AMQConnectionModel amqConnectionModel = connections.next();
+ final String userName = amqConnectionModel.getAuthorizedPrincipal() == null
+ ? ""
+ : amqConnectionModel.getAuthorizedPrincipal().getName();
+ if (userId.equals(userName) && !amqConnectionModel.isSessionNameUnique(name))
{
return false;
}
diff --git a/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java b/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
index 0e6b4d3..29f9fc5 100644
--- a/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
+++ b/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
@@ -78,6 +78,7 @@
import org.apache.qpid.server.txn.TimeoutDtxException;
import org.apache.qpid.server.txn.UnknownDtxBranchException;
import org.apache.qpid.server.util.Action;
+import org.apache.qpid.server.util.Deletable;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.transport.*;
import org.slf4j.Logger;
@@ -88,7 +89,9 @@
public class ServerSession extends Session
implements AuthorizationHolder,
- AMQSessionModel, LogSubject, AsyncAutoCommitTransaction.FutureRecorder
+ AMQSessionModel<ServerSession,ServerConnection>, LogSubject, AsyncAutoCommitTransaction.FutureRecorder,
+ Deletable<ServerSession>
+
{
private static final Logger _logger = LoggerFactory.getLogger(ServerSession.class);
@@ -132,7 +135,7 @@
private Map<String, ConsumerTarget_0_10> _subscriptions = new ConcurrentHashMap<String, ConsumerTarget_0_10>();
- private final List<Action<ServerSession>> _taskList = new CopyOnWriteArrayList<Action<ServerSession>>();
+ private final List<Action<? super ServerSession>> _taskList = new CopyOnWriteArrayList<Action<? super ServerSession>>();
private final TransactionTimeoutHelper _transactionTimeoutHelper;
@@ -374,7 +377,7 @@
}
_messageDispositionListenerMap.clear();
- for (Action<ServerSession> task : _taskList)
+ for (Action<? super ServerSession> task : _taskList)
{
task.performAction(this);
}
@@ -610,12 +613,12 @@
return getConnection().getAuthorizedSubject();
}
- public void addSessionCloseTask(Action<ServerSession> task)
+ public void addDeleteTask(Action<? super ServerSession> task)
{
_taskList.add(task);
}
- public void removeSessionCloseTask(Action<ServerSession> task)
+ public void removeDeleteTask(Action<? super ServerSession> task)
{
_taskList.remove(task);
}
@@ -652,7 +655,7 @@
return _id;
}
- public AMQConnectionModel getConnectionModel()
+ public ServerConnection getConnectionModel()
{
return getConnection();
}
@@ -922,7 +925,7 @@
}
@Override
- public int compareTo(AMQSessionModel o)
+ public int compareTo(ServerSession o)
{
return getId().compareTo(o.getId());
}
diff --git a/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java b/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
index 1bd5053..b0a60be 100644
--- a/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
+++ b/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
@@ -25,6 +25,8 @@
import java.util.UUID;
import org.apache.log4j.Logger;
+import org.apache.qpid.server.model.ExclusivityPolicy;
+import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.store.StoreException;
import org.apache.qpid.server.exchange.AMQUnknownExchangeType;
import org.apache.qpid.server.exchange.Exchange;
@@ -204,47 +206,12 @@
{
exception(session,method,ExecutionErrorCode.NOT_FOUND, "Queue: " + queueName + " not found");
}
- else if(queue.getAuthorizationHolder() != null && queue.getAuthorizationHolder() != session)
- {
- exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
- }
- else if(queue.isExclusive() && queue.getExclusiveOwningSession() != null && queue.getExclusiveOwningSession() != session)
+ else if(!queue.verifySessionAccess((ServerSession)session))
{
exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
}
else
{
- if(queue.isExclusive())
- {
- ServerSession s = (ServerSession) session;
- queue.setExclusiveOwningSession(s);
-
- ((ServerSession) session).addSessionCloseTask(new Action<ServerSession>()
- {
- public void performAction(ServerSession session)
- {
- if(queue.getExclusiveOwningSession() == session)
- {
- queue.setExclusiveOwningSession(null);
- }
- }
- });
-
- if(queue.getAuthorizationHolder() == null)
- {
- queue.setAuthorizationHolder(s);
- ((ServerSession) session).addSessionCloseTask(new Action<ServerSession>()
- {
- public void performAction(ServerSession session)
- {
- if(queue.getAuthorizationHolder() == session)
- {
- queue.setAuthorizationHolder(null);
- }
- }
- });
- }
- }
FlowCreditManager_0_10 creditManager = new WindowCreditManager(0L,0L);
@@ -302,6 +269,10 @@
{
exception(session, method, ExecutionErrorCode.UNAUTHORIZED_ACCESS, e.getMessage());
}
+ catch (MessageSource.ConsumerAccessRefused consumerAccessRefused)
+ {
+ exception(session, method, ExecutionErrorCode.RESOURCE_LOCKED, "Queue has an incompatible exclusivity policy");
+ }
}
}
}
@@ -1197,7 +1168,7 @@
exception(session, method, errorCode, description);
}
- else if (exclusive && (queue.getExclusiveOwningSession() != null && !queue.getExclusiveOwningSession().equals(session)))
+ else if (!queue.verifySessionAccess((ServerSession)session))
{
String description = "Cannot declare queue('" + queueName + "'),"
+ " as exclusive queue with same name "
@@ -1214,7 +1185,6 @@
try
{
- String owner = method.getExclusive() ? ((ServerSession)session).getClientID() : null;
final String alternateExchangeName = method.getAlternateExchange();
@@ -1227,66 +1197,36 @@
final UUID id = UUIDGenerator.generateQueueUUID(queueName, virtualHost.getName());
- final boolean deleteOnNoConsumer = !exclusive && autoDelete;
+ arguments.put(Queue.ID, id);
+ arguments.put(Queue.NAME, queueName);
- queue = virtualHost.createQueue(id, queueName, method.getDurable(), owner,
- autoDelete, exclusive, deleteOnNoConsumer,
- arguments);
+ LifetimePolicy lifetime;
+ if(autoDelete)
+ {
+ lifetime = exclusive ? LifetimePolicy.DELETE_ON_SESSION_END
+ : LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS;
+ }
+ else
+ {
+ lifetime = LifetimePolicy.PERMANENT;
+ }
- if (autoDelete && exclusive)
- {
- final AMQQueue q = queue;
- final Action<ServerSession> deleteQueueTask = new Action<ServerSession>()
- {
- public void performAction(ServerSession session)
- {
- try
- {
- virtualHost.removeQueue(q);
- }
- catch (QpidSecurityException e)
- {
- exception(session, method, ExecutionErrorCode.UNAUTHORIZED_ACCESS, e.getMessage());
- }
- }
- };
- final ServerSession s = (ServerSession) session;
- s.addSessionCloseTask(deleteQueueTask);
- queue.addQueueDeleteTask(new Action<AMQQueue>()
- {
- public void performAction(AMQQueue queue)
- {
- s.removeSessionCloseTask(deleteQueueTask);
- }
- });
- }
- if (exclusive)
- {
- final AMQQueue q = queue;
- final Action<ServerSession> removeExclusive = new Action<ServerSession>()
- {
- public void performAction(ServerSession session)
- {
- q.setAuthorizationHolder(null);
- q.setExclusiveOwningSession(null);
- }
- };
- final ServerSession s = (ServerSession) session;
- q.setExclusiveOwningSession(s);
- s.addSessionCloseTask(removeExclusive);
- queue.addQueueDeleteTask(new Action<AMQQueue>()
- {
- public void performAction(AMQQueue queue)
- {
- s.removeSessionCloseTask(removeExclusive);
- }
- });
- }
+ arguments.put(Queue.LIFETIME_POLICY, lifetime);
+
+ ExclusivityPolicy exclusivityPolicy = exclusive ? ExclusivityPolicy.SESSION : ExclusivityPolicy.NONE;
+
+
+ arguments.put(Queue.DURABLE, method.getDurable());
+
+ arguments.put(Queue.EXCLUSIVE, exclusivityPolicy);
+
+ queue = virtualHost.createQueue((ServerSession)session, arguments);
+
}
catch(QueueExistsException qe)
{
queue = qe.getExistingQueue();
- if (exclusive && (queue.getExclusiveOwningSession() != null && !queue.getExclusiveOwningSession().equals(session)))
+ if (!queue.verifySessionAccess((ServerSession)session))
{
String description = "Cannot declare queue('" + queueName + "'),"
+ " as exclusive queue with same name "
@@ -1347,11 +1287,7 @@
}
else
{
- if(queue.getAuthorizationHolder() != null && queue.getAuthorizationHolder() != session)
- {
- exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
- }
- else if(queue.isExclusive() && queue.getExclusiveOwningSession() != null && queue.getExclusiveOwningSession() != session)
+ if(!queue.verifySessionAccess((ServerSession)session))
{
exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
}
@@ -1424,7 +1360,7 @@
result.setQueue(queue.getName());
result.setDurable(queue.isDurable());
result.setExclusive(queue.isExclusive());
- result.setAutoDelete(queue.isAutoDelete());
+ result.setAutoDelete(queue.getLifetimePolicy() != LifetimePolicy.PERMANENT);
Map<String, Object> arguments = new LinkedHashMap<String, Object>();
Collection<String> availableAttrs = queue.getAvailableAttributes();
@@ -1500,7 +1436,6 @@
public void closed(Session session)
{
setThreadSubject(session);
-
ServerSession serverSession = (ServerSession)session;
serverSession.stopSubscriptions();
diff --git a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
index b15b3f0..fe1cb62 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
@@ -22,6 +22,7 @@
import java.nio.ByteBuffer;
import java.util.*;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
@@ -30,6 +31,8 @@
import org.apache.qpid.AMQConnectionException;
import org.apache.qpid.AMQException;
import org.apache.qpid.server.filter.AMQInvalidArgumentException;
+import org.apache.qpid.server.filter.Filterable;
+import org.apache.qpid.server.filter.MessageFilter;
import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.common.AMQPFilterTypes;
import org.apache.qpid.framing.AMQMethodBody;
@@ -86,7 +89,9 @@
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.transport.TransportException;
-public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.FutureRecorder
+public class AMQChannel<T extends AMQProtocolSession<T>>
+ implements AMQSessionModel<AMQChannel<T>,T>,
+ AsyncAutoCommitTransaction.FutureRecorder
{
public static final int DEFAULT_PREFETCH = 4096;
@@ -140,7 +145,7 @@
private final AtomicLong _txnRejects = new AtomicLong(0);
private final AtomicLong _txnCount = new AtomicLong(0);
- private final AMQProtocolSession _session;
+ private final T _session;
private AtomicBoolean _closing = new AtomicBoolean(false);
private final Set<Object> _blockingEntities = Collections.synchronizedSet(new HashSet<Object>());
@@ -163,12 +168,15 @@
private final TransactionTimeoutHelper _transactionTimeoutHelper;
private final UUID _id = UUID.randomUUID();
+ private final List<Action<? super AMQChannel<T>>> _taskList =
+ new CopyOnWriteArrayList<Action<? super AMQChannel<T>>>();
+
private final CapacityCheckAction _capacityCheckAction = new CapacityCheckAction();
private final ImmediateAction _immediateAction = new ImmediateAction();
- public AMQChannel(AMQProtocolSession session, int channelId, MessageStore messageStore)
+ public AMQChannel(T session, int channelId, MessageStore messageStore)
throws AMQException
{
_session = session;
@@ -526,7 +534,8 @@
public AMQShortString consumeFromSource(AMQShortString tag, MessageSource source, boolean acks,
FieldTable filters, boolean exclusive, boolean noLocal)
throws AMQException, QpidSecurityException, MessageSource.ExistingConsumerPreventsExclusive,
- MessageSource.ExistingExclusiveConsumer, AMQInvalidArgumentException
+ MessageSource.ExistingExclusiveConsumer, AMQInvalidArgumentException,
+ MessageSource.ConsumerAccessRefused
{
if (tag == null)
{
@@ -580,7 +589,15 @@
{
filterManager = new SimpleFilterManager();
}
- filterManager.add(new FilterSupport.NoLocalFilter(source));
+ final Object connectionReference = getConnectionReference();
+ filterManager.add(new MessageFilter()
+ {
+ @Override
+ public boolean matches(final Filterable message)
+ {
+ return message.getConnectionReference() != connectionReference;
+ }
+ });
}
Consumer sub =
source.addConsumer(target,
@@ -609,6 +626,11 @@
_tag2SubscriptionTargetMap.remove(tag);
throw e;
}
+ catch (MessageSource.ConsumerAccessRefused e)
+ {
+ _tag2SubscriptionTargetMap.remove(tag);
+ throw e;
+ }
return tag;
}
@@ -657,6 +679,13 @@
CurrentActor.get().message(_logSubject, operationalLogMessage);
unsubscribeAllConsumers();
+
+ for (Action<? super AMQChannel<T>> task : _taskList)
+ {
+ task.performAction(this);
+ }
+
+
_transaction.rollback();
try
@@ -692,9 +721,10 @@
Consumer sub = me.getValue().getConsumer();
-
- sub.close();
-
+ if(sub != null)
+ {
+ sub.close();
+ }
}
_tag2SubscriptionTargetMap.clear();
@@ -1192,7 +1222,8 @@
return _id;
}
- public AMQConnectionModel getConnectionModel()
+ @Override
+ public T getConnectionModel()
{
return _session;
}
@@ -1208,11 +1239,23 @@
}
@Override
- public int compareTo(AMQSessionModel o)
+ public int compareTo(AMQChannel o)
{
return getId().compareTo(o.getId());
}
+ @Override
+ public void addDeleteTask(final Action<? super AMQChannel<T>> task)
+ {
+ _taskList.add(task);
+ }
+
+ @Override
+ public void removeDeleteTask(final Action<? super AMQChannel<T>> task)
+ {
+ _taskList.remove(task);
+ }
+
private class ImmediateAction<C extends Consumer> implements Action<MessageInstance<?,C>>
{
diff --git a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
index 5e95701..68f1ad7 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
@@ -33,7 +33,6 @@
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
@@ -47,18 +46,15 @@
import org.apache.qpid.AMQConnectionException;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.*;
-import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.codec.AMQCodecFactory;
import org.apache.qpid.common.QpidProperties;
import org.apache.qpid.common.ServerPropertyNames;
import org.apache.qpid.properties.ConnectionStartProperties;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.protocol.AMQMethodEvent;
-import org.apache.qpid.protocol.AMQMethodListener;
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.ServerMessage;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.configuration.BrokerProperties;
import org.apache.qpid.server.protocol.v0_8.handler.ServerMethodDispatcherImpl;
import org.apache.qpid.server.logging.LogActor;
@@ -78,6 +74,7 @@
import org.apache.qpid.server.protocol.v0_8.state.AMQStateManager;
import org.apache.qpid.server.stats.StatisticsCounter;
import org.apache.qpid.server.consumer.Consumer;
+import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -86,7 +83,7 @@
import org.apache.qpid.transport.network.NetworkConnection;
import org.apache.qpid.util.BytesDataOutput;
-public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSession
+public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSession<AMQProtocolEngine>
{
private static final Logger _logger = Logger.getLogger(AMQProtocolEngine.class);
@@ -103,9 +100,11 @@
private VirtualHost _virtualHost;
- private final Map<Integer, AMQChannel> _channelMap = new HashMap<Integer, AMQChannel>();
+ private final Map<Integer, AMQChannel<AMQProtocolEngine>> _channelMap =
+ new HashMap<Integer, AMQChannel<AMQProtocolEngine>>();
- private final AMQChannel[] _cachedChannels = new AMQChannel[CHANNEL_CACHE_SIZE + 1];
+ @SuppressWarnings("unchecked")
+ private final AMQChannel<AMQProtocolEngine>[] _cachedChannels = new AMQChannel[CHANNEL_CACHE_SIZE + 1];
/**
* The channels that the latest call to {@link #received(ByteBuffer)} applied to.
@@ -114,9 +113,8 @@
*
* Thread-safety: guarded by {@link #_receivedLock}.
*/
- private final Set<AMQChannel> _channelsForCurrentMessage = new HashSet<AMQChannel>();
-
- private final CopyOnWriteArraySet<AMQMethodListener> _frameListeners = new CopyOnWriteArraySet<AMQMethodListener>();
+ private final Set<AMQChannel<AMQProtocolEngine>> _channelsForCurrentMessage =
+ new HashSet<AMQChannel<AMQProtocolEngine>>();
private final AMQStateManager _stateManager;
@@ -124,10 +122,6 @@
private SaslServer _saslServer;
- private Object _lastReceived;
-
- private Object _lastSent;
-
private volatile boolean _closed;
// maximum number of channels this session should have
@@ -136,8 +130,8 @@
/* AMQP Version for this session */
private ProtocolVersion _protocolVersion = ProtocolVersion.getLatestSupportedVersion();
private MethodRegistry _methodRegistry = MethodRegistry.getMethodRegistry(_protocolVersion);
- private FieldTable _clientProperties;
- private final List<Task> _taskList = new CopyOnWriteArrayList<Task>();
+ private final List<Action<? super AMQProtocolEngine>> _taskList =
+ new CopyOnWriteArrayList<Action<? super AMQProtocolEngine>>();
private Map<Integer, Long> _closingChannelsList = new ConcurrentHashMap<Integer, Long>();
private ProtocolOutputConverter _protocolOutputConverter;
@@ -153,12 +147,9 @@
private long _lastIoTime;
private long _writtenBytes;
- private long _readBytes;
-
private long _maxFrameSize;
private final AtomicBoolean _closing = new AtomicBoolean(false);
- private long _createTime = System.currentTimeMillis();
private StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
@@ -176,6 +167,7 @@
private volatile boolean _closeWhenNoRoute;
private volatile boolean _stopped;
+ private long _readBytes;
public AMQProtocolEngine(Broker broker,
NetworkConnection network,
@@ -258,15 +250,14 @@
final long arrivalTime = System.currentTimeMillis();
_lastReceivedTime = arrivalTime;
_lastIoTime = arrivalTime;
+ _readBytes += msg.remaining();
_receivedLock.lock();
try
{
final ArrayList<AMQDataBlock> dataBlocks = _codecFactory.getDecoder().decodeBuffer(msg);
- final int len = dataBlocks.size();
- for (int i = 0; i < len; i++)
+ for (AMQDataBlock dataBlock : dataBlocks)
{
- AMQDataBlock dataBlock = dataBlocks.get(i);
try
{
dataBlockReceived(dataBlock);
@@ -316,7 +307,7 @@
private void receivedComplete()
{
- for (AMQChannel channel : _channelsForCurrentMessage)
+ for (AMQChannel<AMQProtocolEngine> channel : _channelsForCurrentMessage)
{
channel.receivedComplete();
}
@@ -334,7 +325,6 @@
*/
private void dataBlockReceived(AMQDataBlock message) throws Exception
{
- _lastReceived = message;
if (message instanceof ProtocolInitiation)
{
protocolInitiationReceived((ProtocolInitiation) message);
@@ -363,7 +353,7 @@
private void frameReceived(AMQFrame frame) throws AMQException
{
int channelId = frame.getChannel();
- AMQChannel amqChannel = _channelMap.get(channelId);
+ AMQChannel<AMQProtocolEngine> amqChannel = _channelMap.get(channelId);
if(amqChannel != null)
{
// The _receivedLock is already acquired in the caller
@@ -558,14 +548,6 @@
{
boolean wasAnyoneInterested = _stateManager.methodReceived(evt);
- if (!_frameListeners.isEmpty())
- {
- for (AMQMethodListener listener : _frameListeners)
- {
- wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested;
- }
- }
-
if (!wasAnyoneInterested)
{
throw new AMQNoMethodHandlerException(evt);
@@ -611,11 +593,6 @@
}
catch (Exception e)
{
- for (AMQMethodListener listener : _frameListeners)
- {
- listener.error(e);
- }
-
_logger.error("Unexpected exception while processing frame. Closing connection.", e);
closeProtocolSession();
@@ -625,7 +602,7 @@
public void contentHeaderReceived(int channelId, ContentHeaderBody body) throws AMQException
{
- AMQChannel channel = getAndAssertChannel(channelId);
+ AMQChannel<AMQProtocolEngine> channel = getAndAssertChannel(channelId);
channel.publishContentHeader(body);
@@ -633,7 +610,7 @@
public void contentBodyReceived(int channelId, ContentBody body) throws AMQException
{
- AMQChannel channel = getAndAssertChannel(channelId);
+ AMQChannel<AMQProtocolEngine> channel = getAndAssertChannel(channelId);
channel.publishContentBody(body);
}
@@ -681,17 +658,17 @@
_contextKey = contextKey;
}
- public List<AMQChannel> getChannels()
+ public List<AMQChannel<AMQProtocolEngine>> getChannels()
{
synchronized (_channelMap)
{
- return new ArrayList<AMQChannel>(_channelMap.values());
+ return new ArrayList<AMQChannel<AMQProtocolEngine>>(_channelMap.values());
}
}
- public AMQChannel getAndAssertChannel(int channelId) throws AMQException
+ public AMQChannel<AMQProtocolEngine> getAndAssertChannel(int channelId) throws AMQException
{
- AMQChannel channel = getChannel(channelId);
+ AMQChannel<AMQProtocolEngine> channel = getChannel(channelId);
if (channel == null)
{
throw new AMQException(AMQConstant.NOT_FOUND, "Channel not found with id:" + channelId);
@@ -700,9 +677,9 @@
return channel;
}
- public AMQChannel getChannel(int channelId)
+ public AMQChannel<AMQProtocolEngine> getChannel(int channelId)
{
- final AMQChannel channel =
+ final AMQChannel<AMQProtocolEngine> channel =
((channelId & CHANNEL_CACHE_SIZE) == channelId) ? _cachedChannels[channelId] : _channelMap.get(channelId);
if ((channel == null) || channel.isClosing())
{
@@ -719,7 +696,7 @@
return !_closingChannelsList.isEmpty() && _closingChannelsList.containsKey(channelId);
}
- public void addChannel(AMQChannel channel) throws AMQException
+ public void addChannel(AMQChannel<AMQProtocolEngine> channel) throws AMQException
{
if (_closed)
{
@@ -770,7 +747,7 @@
_maxNoOfChannels = value;
}
- public void commitTransactions(AMQChannel channel) throws AMQException
+ public void commitTransactions(AMQChannel<AMQProtocolEngine> channel) throws AMQException
{
if ((channel != null) && channel.isTransactional())
{
@@ -778,7 +755,7 @@
}
}
- public void rollbackTransactions(AMQChannel channel) throws AMQException
+ public void rollbackTransactions(AMQChannel<AMQProtocolEngine> channel) throws AMQException
{
if ((channel != null) && channel.isTransactional())
{
@@ -802,7 +779,7 @@
public void closeChannel(int channelId, AMQConstant cause, String message)
{
- final AMQChannel channel = getChannel(channelId);
+ final AMQChannel<AMQProtocolEngine> channel = getChannel(channelId);
if (channel == null)
{
throw new IllegalArgumentException("Unknown channel id");
@@ -879,12 +856,10 @@
/**
* Closes all channels that were opened by this protocol session. This frees up all resources used by the channel.
- *
- * @throws AMQException if an error occurs while closing any channel
*/
private void closeAllChannels()
{
- for (AMQChannel channel : getChannels())
+ for (AMQChannel<AMQProtocolEngine> channel : getChannels())
{
channel.close();
}
@@ -929,9 +904,9 @@
closeAllChannels();
- for (Task task : _taskList)
+ for (Action<? super AMQProtocolEngine> task : _taskList)
{
- task.doTask(this);
+ task.performAction(this);
}
synchronized(this)
@@ -961,7 +936,7 @@
}
catch (InterruptedException e)
{
-
+ // do nothing
}
finally
{
@@ -1027,11 +1002,6 @@
return getRemoteAddress() + "(" + (getAuthorizedPrincipal() == null ? "?" : getAuthorizedPrincipal().getName() + ")");
}
- public String dump()
- {
- return this + " last_sent=" + _lastSent + " last_received=" + _lastReceived;
- }
-
/** @return an object that can be used to identity */
public Object getKey()
{
@@ -1069,10 +1039,9 @@
public void setClientProperties(FieldTable clientProperties)
{
- _clientProperties = clientProperties;
- if (_clientProperties != null)
+ if (clientProperties != null)
{
- String closeWhenNoRoute = _clientProperties.getString(ConnectionStartProperties.QPID_CLOSE_WHEN_NO_ROUTE);
+ String closeWhenNoRoute = clientProperties.getString(ConnectionStartProperties.QPID_CLOSE_WHEN_NO_ROUTE);
if (closeWhenNoRoute != null)
{
_closeWhenNoRoute = Boolean.parseBoolean(closeWhenNoRoute);
@@ -1082,10 +1051,10 @@
}
}
- _clientVersion = _clientProperties.getString(ConnectionStartProperties.VERSION_0_8);
- _clientProduct = _clientProperties.getString(ConnectionStartProperties.PRODUCT);
+ _clientVersion = clientProperties.getString(ConnectionStartProperties.VERSION_0_8);
+ _clientProduct = clientProperties.getString(ConnectionStartProperties.PRODUCT);
- String clientId = _clientProperties.getString(ConnectionStartProperties.CLIENT_ID_0_8);
+ String clientId = clientProperties.getString(ConnectionStartProperties.CLIENT_ID_0_8);
if (clientId != null)
{
setContextKey(new AMQShortString(clientId));
@@ -1118,11 +1087,6 @@
return _protocolVersion.getMinorVersion();
}
- public boolean isProtocolVersion(byte major, byte minor)
- {
- return (getProtocolMajorVersion() == major) && (getProtocolMinorVersion() == minor);
- }
-
public MethodRegistry getRegistry()
{
return getMethodRegistry();
@@ -1141,12 +1105,12 @@
}
- public void addSessionCloseTask(Task task)
+ public void addDeleteTask(Action<? super AMQProtocolEngine> task)
{
_taskList.add(task);
}
- public void removeSessionCloseTask(Task task)
+ public void removeDeleteTask(Action<? super AMQProtocolEngine> task)
{
_taskList.remove(task);
}
@@ -1341,51 +1305,11 @@
return _clientProduct;
}
- public String getPrincipalAsString()
- {
- return getAuthId();
- }
-
public long getSessionCountLimit()
{
return getMaximumNumberOfChannels();
}
- public Boolean isIncoming()
- {
- return true;
- }
-
- public Boolean isSystemConnection()
- {
- return false;
- }
-
- public Boolean isFederationLink()
- {
- return false;
- }
-
- public String getAuthId()
- {
- return getAuthorizedPrincipal() == null ? null : getAuthorizedPrincipal().getName();
- }
-
- public Integer getRemotePID()
- {
- return null;
- }
-
- public String getRemoteProcessName()
- {
- return null;
- }
-
- public Integer getRemoteParentPID()
- {
- return null;
- }
-
public boolean isDurable()
{
return false;
@@ -1401,52 +1325,6 @@
return String.valueOf(getRemoteAddress());
}
- public long getCreateTime()
- {
- return _createTime;
- }
-
- public Boolean isShadow()
- {
- return false;
- }
-
- public void mgmtClose()
- {
- MethodRegistry methodRegistry = getMethodRegistry();
- ConnectionCloseBody responseBody =
- methodRegistry.createConnectionCloseBody(
- AMQConstant.REPLY_SUCCESS.getCode(),
- new AMQShortString("The connection was closed using the broker's management interface."),
- 0,0);
-
- // This seems ugly but because we use closeConnection in both normal
- // broker operation and as part of the management interface it cannot
- // be avoided. The Current Actor will be null when this method is
- // called via the QMF management interface. As such we need to set one.
- boolean removeActor = false;
- if (CurrentActor.get() == null)
- {
- removeActor = true;
- CurrentActor.set(new ManagementActor(_actor.getRootMessageLogger()));
- }
-
- try
- {
- writeFrame(responseBody.generateFrame(0));
-
- closeSession();
-
- }
- finally
- {
- if (removeActor)
- {
- CurrentActor.remove();
- }
- }
- }
-
public void mgmtCloseChannel(int channelId)
{
MethodRegistry methodRegistry = getMethodRegistry();
@@ -1481,14 +1359,9 @@
}
}
- public String getClientID()
+ public void closeSession(AMQChannel<AMQProtocolEngine> session, AMQConstant cause, String message)
{
- return getContextKey().toString();
- }
-
- public void closeSession(AMQSessionModel session, AMQConstant cause, String message)
- {
- int channelId = ((AMQChannel)session).getChannelId();
+ int channelId = session.getChannelId();
closeChannel(channelId, cause, message);
MethodRegistry methodRegistry = getMethodRegistry();
@@ -1506,7 +1379,7 @@
closeConnection(0, new AMQConnectionException(cause, message, 0, 0,
getProtocolOutputConverter().getProtocolMajorVersion(),
getProtocolOutputConverter().getProtocolMinorVersion(),
- (Throwable) null));
+ null));
}
public void block()
@@ -1516,7 +1389,7 @@
if(!_blocking)
{
_blocking = true;
- for(AMQChannel channel : _channelMap.values())
+ for(AMQChannel<AMQProtocolEngine> channel : _channelMap.values())
{
channel.block();
}
@@ -1531,7 +1404,7 @@
if(_blocking)
{
_blocking = false;
- for(AMQChannel channel : _channelMap.values())
+ for(AMQChannel<AMQProtocolEngine> channel : _channelMap.values())
{
channel.unblock();
}
@@ -1544,9 +1417,9 @@
return _closed;
}
- public List<AMQSessionModel> getSessionModels()
+ public List<AMQChannel<AMQProtocolEngine>> getSessionModels()
{
- return new ArrayList<AMQSessionModel>(getChannels());
+ return new ArrayList<AMQChannel<AMQProtocolEngine>>(getChannels());
}
public LogSubject getLogSubject()
@@ -1620,16 +1493,17 @@
return String.valueOf(getContextKey());
}
+ @Override
+ public String getRemoteContainerName()
+ {
+ return String.valueOf(getContextKey());
+ }
+
public void setDeferFlush(boolean deferFlush)
{
_deferFlush = deferFlush;
}
- public String getUserName()
- {
- return getAuthorizedPrincipal().getName();
- }
-
public final class WriteDeliverMethod
implements ClientDeliveryMethod
{
diff --git a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolSession.java b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolSession.java
index 58a3b5d..045367b 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolSession.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolSession.java
@@ -37,12 +37,14 @@
import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
import org.apache.qpid.server.logging.LogActor;
import org.apache.qpid.server.protocol.AMQConnectionModel;
+import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.v0_8.output.ProtocolOutputConverter;
import org.apache.qpid.server.security.AuthorizationHolder;
import org.apache.qpid.server.virtualhost.VirtualHost;
-public interface AMQProtocolSession extends AMQVersionAwareProtocolSession, AuthorizationHolder, AMQConnectionModel
+public interface AMQProtocolSession<T extends AMQProtocolSession<T>>
+ extends AMQVersionAwareProtocolSession, AuthorizationHolder, AMQConnectionModel<T,AMQChannel<T>>
{
long getSessionID();
@@ -69,11 +71,6 @@
*/
SocketAddress getLocalAddress();
- public static interface Task
- {
- public void doTask(AMQProtocolSession session);
- }
-
/**
* Get the context key associated with this session. Context key is described in the AMQ protocol specification (RFC
* 6).
@@ -98,7 +95,7 @@
*
* @return null if no channel exists, the channel otherwise
*/
- AMQChannel getChannel(int channelId);
+ AMQChannel<T> getChannel(int channelId);
/**
* Associate a channel with this session.
@@ -106,7 +103,7 @@
* @param channel the channel to associate with this session. It is an error to associate the same channel with more
* than one session but this is not validated.
*/
- void addChannel(AMQChannel channel) throws AMQException;
+ void addChannel(AMQChannel<T> channel) throws AMQException;
/**
* Close a specific channel. This will remove any resources used by the channel, including: <ul><li>any queue
@@ -185,10 +182,6 @@
void setVirtualHost(VirtualHost virtualHost) throws AMQException;
- void addSessionCloseTask(Task task);
-
- void removeSessionCloseTask(Task task);
-
public ProtocolOutputConverter getProtocolOutputConverter();
void setAuthorizedSubject(Subject authorizedSubject);
@@ -209,11 +202,11 @@
void setMaximumNumberOfChannels(Long value);
- void commitTransactions(AMQChannel channel) throws AMQException;
+ void commitTransactions(AMQChannel<T> channel) throws AMQException;
- void rollbackTransactions(AMQChannel channel) throws AMQException;
+ void rollbackTransactions(AMQChannel<T> channel) throws AMQException;
- List<AMQChannel> getChannels();
+ List<AMQChannel<T>> getChannels();
void mgmtCloseChannel(int channelId);
diff --git a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicConsumeMethodHandler.java b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicConsumeMethodHandler.java
index ce90de7..c93c164 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicConsumeMethodHandler.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicConsumeMethodHandler.java
@@ -99,16 +99,6 @@
{
final AMQShortString consumerTagName;
- if (queue.isExclusive() && !queue.isDurable())
- {
- AMQSessionModel session = queue.getExclusiveOwningSession();
- if (session == null || session.getConnectionModel() != protocolConnection)
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue " + queue.getName() + " is exclusive, but not created on this Connection.");
- }
- }
-
if (body.getConsumerTag() != null)
{
consumerTagName = body.getConsumerTag().intern(false);
@@ -184,6 +174,13 @@
+ queue.getName()
+ " permission denied");
}
+ catch (MessageSource.ConsumerAccessRefused consumerAccessRefused)
+ {
+ throw body.getChannelException(AMQConstant.ACCESS_REFUSED,
+ "Cannot subscribe to queue "
+ + queue.getName()
+ + " as it already has an incompatible exclusivity policy");
+ }
}
}
diff --git a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicGetMethodHandler.java b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicGetMethodHandler.java
index c9a7cc6..4370004 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicGetMethodHandler.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicGetMethodHandler.java
@@ -98,15 +98,6 @@
}
else
{
- if (queue.isExclusive())
- {
- AMQSessionModel session = queue.getExclusiveOwningSession();
- if (session == null || session.getConnectionModel() != protocolConnection)
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue is exclusive, but not created on this Connection.");
- }
- }
try
{
@@ -136,6 +127,11 @@
"The GET request has been evaluated as an exclusive consumer, " +
"this is likely due to a programming error in the Qpid broker");
}
+ catch (MessageSource.ConsumerAccessRefused consumerAccessRefused)
+ {
+ throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
+ "Queue has an incompatible exclusivit policy");
+ }
}
}
}
@@ -145,7 +141,7 @@
final AMQChannel channel,
final boolean acks)
throws AMQException, QpidSecurityException, MessageSource.ExistingConsumerPreventsExclusive,
- MessageSource.ExistingExclusiveConsumer
+ MessageSource.ExistingExclusiveConsumer, MessageSource.ConsumerAccessRefused
{
final FlowCreditManager singleMessageCredit = new MessageOnlyCreditManager(1L);
diff --git a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueBindHandler.java b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueBindHandler.java
index 401718d..9b875cc 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueBindHandler.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueBindHandler.java
@@ -116,15 +116,6 @@
try
{
- if (queue.isExclusive() && !queue.isDurable())
- {
- AMQSessionModel session = queue.getExclusiveOwningSession();
- if (session == null || session.getConnectionModel() != protocolConnection)
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue " + queue.getName() + " is exclusive, but not created on this Connection.");
- }
- }
Map<String,Object> arguments = FieldTable.convertToMap(body.getArguments());
String bindingKey = String.valueOf(routingKey);
@@ -144,10 +135,6 @@
}
}
}
- catch (AMQException e)
- {
- throw body.getChannelException(AMQConstant.CHANNEL_ERROR, e.toString());
- }
catch (QpidSecurityException e)
{
throw body.getConnectionException(AMQConstant.ACCESS_REFUSED, e.getMessage());
diff --git a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeclareHandler.java b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeclareHandler.java
index 5b55256..215e3f2 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeclareHandler.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeclareHandler.java
@@ -29,6 +29,9 @@
import org.apache.qpid.framing.QueueDeclareBody;
import org.apache.qpid.framing.QueueDeclareOkBody;
import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.server.model.ExclusivityPolicy;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.protocol.v0_8.AMQChannel;
import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.protocol.v0_8.AMQProtocolSession;
@@ -39,7 +42,6 @@
import org.apache.qpid.server.protocol.v0_8.state.StateAwareMethodListener;
import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.server.util.Action;
-import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Map;
@@ -96,9 +98,7 @@
}
else
{
- AMQSessionModel owningSession = queue.getExclusiveOwningSession();
- if (queue.isExclusive() && !queue.isDurable()
- && (owningSession == null || owningSession.getConnectionModel() != protocolConnection))
+ if (!queue.verifySessionAccess(channel))
{
throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
"Queue " + queue.getName() + " is exclusive, but not created on this Connection.");
@@ -114,42 +114,15 @@
try
{
- queue = createQueue(queueName, body, virtualHost, protocolConnection);
- queue.setAuthorizationHolder(protocolConnection);
-
- if (body.getExclusive())
- {
- queue.setExclusiveOwningSession(protocolConnection.getChannel(channelId));
- queue.setAuthorizationHolder(protocolConnection);
-
- if(!body.getDurable())
- {
- final AMQQueue q = queue;
- final AMQProtocolSession.Task sessionCloseTask = new AMQProtocolSession.Task()
- {
- public void doTask(AMQProtocolSession session)
- {
- q.setExclusiveOwningSession(null);
- }
- };
- protocolConnection.addSessionCloseTask(sessionCloseTask);
- queue.addQueueDeleteTask(new Action<AMQQueue>() {
- public void performAction(AMQQueue queue)
- {
- protocolConnection.removeSessionCloseTask(sessionCloseTask);
- }
- });
- }
- }
+ queue = createQueue(channel, queueName, body, virtualHost, protocolConnection);
}
catch(QueueExistsException qe)
{
queue = qe.getExistingQueue();
- AMQSessionModel owningSession = queue.getExclusiveOwningSession();
- if (queue.isExclusive() && !queue.isDurable() && (owningSession == null || owningSession.getConnectionModel() != protocolConnection))
+ if (!queue.verifySessionAccess(channel))
{
throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
"Queue " + queue.getName() + " is exclusive, but not created on this Connection.");
@@ -161,19 +134,12 @@
"Cannot re-declare queue '" + queue.getName() + "' with different exclusivity (was: "
+ queue.isExclusive() + " requested " + body.getExclusive() + ")");
}
- else if (body.getExclusive() && !(queue.isDurable() ? String.valueOf(queue.getOwner()).equals(session.getClientID()) : (owningSession == null || owningSession.getConnectionModel() == protocolConnection)))
- {
- throw body.getChannelException(AMQConstant.ALREADY_EXISTS, "Cannot declare queue('" + queueName + "'), "
- + "as exclusive queue with same name "
- + "declared on another client ID('"
- + queue.getOwner() + "') your clientID('" + session.getClientID() + "')");
-
- }
- else if(queue.isAutoDelete() != body.getAutoDelete())
+ else if((body.getAutoDelete() && queue.getLifetimePolicy() != LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS)
+ || (!body.getAutoDelete() && queue.getLifetimePolicy() != ((body.getExclusive() && !body.getDurable()) ? LifetimePolicy.DELETE_ON_CONNECTION_CLOSE : LifetimePolicy.PERMANENT)))
{
throw body.getChannelException(AMQConstant.ALREADY_EXISTS,
- "Cannot re-declare queue '" + queue.getName() + "' with different auto-delete (was: "
- + queue.isAutoDelete() + " requested " + body.getAutoDelete() + ")");
+ "Cannot re-declare queue '" + queue.getName() + "' with different lifetime policy (was: "
+ + queue.getLifetimePolicy() + " requested autodelete: " + body.getAutoDelete() + ")");
}
else if(queue.isDurable() != body.getDurable())
{
@@ -211,7 +177,7 @@
return new AMQShortString("tmp_" + UUID.randomUUID());
}
- protected AMQQueue createQueue(final AMQShortString queueName,
+ protected AMQQueue createQueue(final AMQChannel channel, final AMQShortString queueName,
QueueDeclareBody body,
final VirtualHost virtualHost,
final AMQProtocolSession session)
@@ -222,47 +188,35 @@
final boolean autoDelete = body.getAutoDelete();
final boolean exclusive = body.getExclusive();
- String owner = exclusive ? AMQShortString.toString(session.getContextKey()) : null;
- Map<String, Object> arguments =
+ Map<String, Object> attributes =
QueueArgumentsConverter.convertWireArgsToModel(FieldTable.convertToMap(body.getArguments()));
- String queueNameString = AMQShortString.toString(queueName);
- final UUID id = UUIDGenerator.generateQueueUUID(queueNameString, virtualHost.getName());
+ final String queueNameString = AMQShortString.toString(queueName);
+ attributes.put(Queue.NAME, queueNameString);
+ attributes.put(Queue.ID, UUIDGenerator.generateQueueUUID(queueNameString, virtualHost.getName()));
+ attributes.put(Queue.DURABLE, durable);
- final AMQQueue queue = virtualHost.createQueue(id, queueNameString, durable, owner, autoDelete,
- exclusive, autoDelete, arguments);
+ LifetimePolicy lifetimePolicy;
+ ExclusivityPolicy exclusivityPolicy;
- if (exclusive && !durable)
+ if(exclusive)
{
- final AMQProtocolSession.Task deleteQueueTask =
- new AMQProtocolSession.Task()
- {
- public void doTask(AMQProtocolSession session)
- {
- if (virtualHost.getQueue(queueName.toString()) == queue)
- {
- try
- {
- virtualHost.removeQueue(queue);
- }
- catch (QpidSecurityException e)
- {
- throw new ConnectionScopedRuntimeException("Permission exception: Unable to remove a temporary queue created by a session which has now removed itself", e);
- }
- }
- }
- };
-
- session.addSessionCloseTask(deleteQueueTask);
-
- queue.addQueueDeleteTask(new Action<AMQQueue>()
- {
- public void performAction(AMQQueue queue)
- {
- session.removeSessionCloseTask(deleteQueueTask);
- }
- });
+ lifetimePolicy = autoDelete
+ ? LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS
+ : durable ? LifetimePolicy.PERMANENT : LifetimePolicy.DELETE_ON_CONNECTION_CLOSE;
+ exclusivityPolicy = durable ? ExclusivityPolicy.CONTAINER : ExclusivityPolicy.CONNECTION;
}
+ else
+ {
+ lifetimePolicy = autoDelete ? LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS : LifetimePolicy.PERMANENT;
+ exclusivityPolicy = ExclusivityPolicy.NONE;
+ }
+
+ attributes.put(Queue.EXCLUSIVE, exclusivityPolicy);
+ attributes.put(Queue.LIFETIME_POLICY, lifetimePolicy);
+
+
+ final AMQQueue queue = virtualHost.createQueue(channel, attributes);
return queue;
}
diff --git a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeleteHandler.java b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeleteHandler.java
index 3a9a6dc..c939e49 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeleteHandler.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueueDeleteHandler.java
@@ -105,8 +105,7 @@
}
else
{
- AMQSessionModel session = queue.getExclusiveOwningSession();
- if (queue.isExclusive() && !queue.isDurable() && (session == null || session.getConnectionModel() != protocolConnection))
+ if (!queue.verifySessionAccess(channel))
{
throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
"Queue " + queue.getName() + " is exclusive, but not created on this Connection.");
diff --git a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueuePurgeHandler.java b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueuePurgeHandler.java
index 6d8f8e6..9d035a3 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueuePurgeHandler.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/QueuePurgeHandler.java
@@ -96,9 +96,7 @@
}
else
{
- AMQSessionModel session = queue.getExclusiveOwningSession();
-
- if (queue.isExclusive() && (session == null || session.getConnectionModel() != protocolConnection))
+ if (!queue.verifySessionAccess(channel))
{
throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
"Queue is exclusive, but not created on this Connection.");
diff --git a/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java b/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java
index 7661d98..f8888f9 100644
--- a/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java
+++ b/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java
@@ -223,7 +223,7 @@
// Then the AMQMinaProtocolSession can join on the returning future without a NPE.
}
- public void closeSession(AMQSessionModel session, AMQConstant cause, String message)
+ public void closeSession(AMQChannel session, AMQConstant cause, String message)
{
super.closeSession(session, cause, message);
diff --git a/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java b/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
index cae61f9..7f8237c 100644
--- a/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
+++ b/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.server.protocol.v1_0;
+import java.security.Principal;
import java.text.MessageFormat;
import java.util.Collection;
import org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint;
@@ -42,7 +43,7 @@
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CONNECTION_FORMAT;
-public class Connection_1_0 implements ConnectionEventListener
+public class Connection_1_0 implements ConnectionEventListener, AMQConnectionModel<Connection_1_0,Session_1_0>
{
private final Port _port;
@@ -53,8 +54,33 @@
private final Collection<Session_1_0> _sessions = Collections.synchronizedCollection(new ArrayList<Session_1_0>());
private final Object _reference = new Object();
- private List<Action<Connection_1_0>> _closeTasks =
- Collections.synchronizedList(new ArrayList<Action<Connection_1_0>>());
+
+ private StatisticsCounter _messageDeliveryStatistics = new StatisticsCounter();
+ private StatisticsCounter _messageReceiptStatistics = new StatisticsCounter();
+ private StatisticsCounter _dataDeliveryStatistics = new StatisticsCounter();
+ private StatisticsCounter _dataReceiptStatistics = new StatisticsCounter();
+
+ private final LogSubject _logSubject = new LogSubject()
+ {
+ @Override
+ public String toLogString()
+ {
+ return "[" +
+ MessageFormat.format(CONNECTION_FORMAT,
+ getConnectionId(),
+ getClientId(),
+ getRemoteAddressString(),
+ _vhost.getName())
+ + "] ";
+
+ }
+ };
+
+ private volatile boolean _stopped;
+
+
+ private List<Action<? super Connection_1_0>> _closeTasks =
+ Collections.synchronizedList(new ArrayList<Action<? super Connection_1_0>>());
@@ -69,7 +95,7 @@
_transport = transport;
_conn = conn;
_connectionId = connectionId;
- _vhost.getConnectionRegistry().registerConnection(_model);
+ _vhost.getConnectionRegistry().registerConnection(this);
}
@@ -80,7 +106,7 @@
public void remoteSessionCreation(SessionEndpoint endpoint)
{
- Session_1_0 session = new Session_1_0(_vhost, this);
+ Session_1_0 session = new Session_1_0(_vhost, this, endpoint);
_sessions.add(session);
endpoint.setSessionEventListener(session);
}
@@ -90,24 +116,24 @@
_sessions.remove(session);
}
- void removeConnectionCloseTask(final Action<Connection_1_0> task)
+ public void removeDeleteTask(final Action<? super Connection_1_0> task)
{
_closeTasks.remove( task );
}
- void addConnectionCloseTask(final Action<Connection_1_0> task)
+ public void addDeleteTask(final Action<? super Connection_1_0> task)
{
_closeTasks.add( task );
}
public void closeReceived()
{
- List<Action<Connection_1_0>> taskCopy;
+ List<Action<? super Connection_1_0>> taskCopy;
synchronized (_closeTasks)
{
- taskCopy = new ArrayList<Action<Connection_1_0>>(_closeTasks);
+ taskCopy = new ArrayList<Action<? super Connection_1_0>>(_closeTasks);
}
- for(Action<Connection_1_0> task : taskCopy)
+ for(Action<? super Connection_1_0> task : taskCopy)
{
task.performAction(this);
}
@@ -115,7 +141,7 @@
{
_closeTasks.clear();
}
- _vhost.getConnectionRegistry().deregisterConnection(_model);
+ _vhost.getConnectionRegistry().deregisterConnection(this);
}
@@ -125,30 +151,6 @@
closeReceived();
}
- private final AMQConnectionModel _model = new AMQConnectionModel()
- {
- private StatisticsCounter _messageDeliveryStatistics = new StatisticsCounter();
- private StatisticsCounter _messageReceiptStatistics = new StatisticsCounter();
- private StatisticsCounter _dataDeliveryStatistics = new StatisticsCounter();
- private StatisticsCounter _dataReceiptStatistics = new StatisticsCounter();
-
- private final LogSubject _logSubject = new LogSubject()
- {
- @Override
- public String toLogString()
- {
- return "[" +
- MessageFormat.format(CONNECTION_FORMAT,
- getConnectionId(),
- getClientId(),
- getRemoteAddressString(),
- _vhost.getName())
- + "] ";
-
- }
- };
-
- private volatile boolean _stopped;
@Override
public void close(AMQConstant cause, String message)
@@ -169,9 +171,9 @@
}
@Override
- public void closeSession(AMQSessionModel session, AMQConstant cause, String message)
+ public void closeSession(Session_1_0 session, AMQConstant cause, String message)
{
- // TODO
+ session.close(cause, message);
}
@Override
@@ -181,9 +183,9 @@
}
@Override
- public List<AMQSessionModel> getSessionModels()
+ public List<Session_1_0> getSessionModels()
{
- return new ArrayList<AMQSessionModel>(_sessions);
+ return new ArrayList<Session_1_0>(_sessions);
}
@Override
@@ -193,12 +195,6 @@
}
@Override
- public String getUserName()
- {
- return getPrincipalAsString();
- }
-
- @Override
public boolean isSessionNameUnique(byte[] name)
{
return true; // TODO
@@ -216,7 +212,13 @@
return _conn.getRemoteContainerId();
}
- @Override
+ @Override
+ public String getRemoteContainerName()
+ {
+ return _conn.getRemoteContainerId();
+ }
+
+ @Override
public String getClientVersion()
{
return ""; //TODO
@@ -228,10 +230,9 @@
return ""; //TODO
}
- @Override
- public String getPrincipalAsString()
+ public Principal getAuthorizedPrincipal()
{
- return String.valueOf(_conn.getUser());
+ return _conn.getUser();
}
@Override
@@ -337,11 +338,10 @@
}
- };
AMQConnectionModel getModel()
{
- return _model;
+ return this;
}
diff --git a/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java b/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
index 546cc79..f7e2d2d 100644
--- a/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
+++ b/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
@@ -30,6 +30,9 @@
import java.util.concurrent.ConcurrentHashMap;
import org.apache.log4j.Logger;
+import org.apache.qpid.server.model.ExclusivityPolicy;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.amqp_1_0.transport.DeliveryStateHandler;
import org.apache.qpid.amqp_1_0.transport.LinkEndpoint;
@@ -207,15 +210,14 @@
if(queue == null)
{
- queue = _vhost.createQueue(
- UUIDGenerator.generateQueueUUID(name, _vhost.getName()),
- name,
- isDurable,
- null,
- true,
- true,
- true,
- Collections.EMPTY_MAP);
+ Map<String,Object> attributes = new HashMap<String,Object>();
+ attributes.put(Queue.ID, UUIDGenerator.generateQueueUUID(name, _vhost.getName()));
+ attributes.put(Queue.NAME, name);
+ attributes.put(Queue.DURABLE, isDurable);
+ attributes.put(Queue.LIFETIME_POLICY, LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS);
+ attributes.put(Queue.EXCLUSIVE, ExclusivityPolicy.LINK);
+
+ queue = _vhost.createQueue(getSession(), attributes);
}
else
{
@@ -308,44 +310,6 @@
exchange.addBinding(binding, queue,null);
source.setDistributionMode(StdDistMode.COPY);
- if(!isDurable)
- {
- final String queueName = name;
- final AMQQueue tempQueue = queue;
-
- final Action<Connection_1_0> deleteQueueTask =
- new Action<Connection_1_0>()
- {
- public void performAction(Connection_1_0 session)
- {
- if (_vhost.getQueue(queueName) == tempQueue)
- {
- try
- {
- _vhost.removeQueue(tempQueue);
- }
- catch (QpidSecurityException e)
- {
- //TODO
- _logger.error("Error removing queue", e);
- }
- }
- }
- };
-
- getSession().getConnection().addConnectionCloseTask(deleteQueueTask);
-
- queue.addQueueDeleteTask(new Action<AMQQueue>()
- {
- public void performAction(AMQQueue queue)
- {
- getSession().getConnection().removeConnectionCloseTask(deleteQueueTask);
- }
-
-
- });
- }
-
qd = new QueueDestination(queue);
}
catch (QpidSecurityException e)
@@ -409,6 +373,11 @@
_logger.info("Cannot add an exclusive consumer to the destination as there is already a consumer");
throw new ConnectionScopedRuntimeException(e);
}
+ catch (MessageSource.ConsumerAccessRefused e)
+ {
+ _logger.info("Cannot add an exclusive consumer to the destination as there is an incompatible exclusivity policy");
+ throw new ConnectionScopedRuntimeException(e);
+ }
}
}
diff --git a/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
index c055d1e..6840c73 100644
--- a/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
+++ b/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
@@ -26,8 +26,10 @@
import org.apache.qpid.amqp_1_0.transport.LinkEndpoint;
import org.apache.qpid.amqp_1_0.transport.ReceivingLinkEndpoint;
import org.apache.qpid.amqp_1_0.transport.SendingLinkEndpoint;
+import org.apache.qpid.amqp_1_0.transport.SessionEndpoint;
import org.apache.qpid.amqp_1_0.transport.SessionEventListener;
import org.apache.qpid.amqp_1_0.type.*;
+import org.apache.qpid.amqp_1_0.type.LifetimePolicy;
import org.apache.qpid.amqp_1_0.type.messaging.*;
import org.apache.qpid.amqp_1_0.type.messaging.Source;
import org.apache.qpid.amqp_1_0.type.messaging.Target;
@@ -36,13 +38,13 @@
import org.apache.qpid.amqp_1_0.type.transport.*;
import org.apache.qpid.amqp_1_0.type.transport.Error;
+import org.apache.qpid.server.model.*;
import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageSource;
-import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.LinkRegistry;
@@ -55,25 +57,34 @@
import org.apache.qpid.server.virtualhost.QueueExistsException;
import java.util.*;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CHANNEL_FORMAT;
-public class Session_1_0 implements SessionEventListener, AMQSessionModel, LogSubject
+public class Session_1_0 implements SessionEventListener, AMQSessionModel<Session_1_0, Connection_1_0>, LogSubject
{
private static final Logger _logger = Logger.getLogger(Session_1_0.class);
private static final Symbol LIFETIME_POLICY = Symbol.valueOf("lifetime-policy");
+ private final SessionEndpoint _endpoint;
private VirtualHost _vhost;
private AutoCommitTransaction _transaction;
private final LinkedHashMap<Integer, ServerTransaction> _openTransactions =
new LinkedHashMap<Integer, ServerTransaction>();
+
+ private final CopyOnWriteArrayList<Action<? super Session_1_0>> _taskList =
+ new CopyOnWriteArrayList<Action<? super Session_1_0>>();
+
private final Connection_1_0 _connection;
private UUID _id = UUID.randomUUID();
+ private AtomicBoolean _closed = new AtomicBoolean();
- public Session_1_0(VirtualHost vhost, final Connection_1_0 connection)
+ public Session_1_0(VirtualHost vhost, final Connection_1_0 connection, final SessionEndpoint endpoint)
{
_vhost = vhost;
+ _endpoint = endpoint;
_transaction = new AutoCommitTransaction(vhost.getMessageStore());
_connection = connection;
@@ -333,64 +344,41 @@
LifetimePolicy lifetimePolicy = properties == null
? null
: (LifetimePolicy) properties.get(LIFETIME_POLICY);
+ Map<String,Object> attributes = new HashMap<String,Object>();
+ attributes.put(org.apache.qpid.server.model.Queue.ID, UUIDGenerator.generateQueueUUID(queueName, _vhost.getName()));
+ attributes.put(org.apache.qpid.server.model.Queue.NAME, queueName);
+ attributes.put(org.apache.qpid.server.model.Queue.DURABLE, false);
- final AMQQueue tempQueue = queue = _vhost.createQueue( UUIDGenerator.generateQueueUUID(queueName, _vhost.getName()),
- queueName,
- false, // durable
- null, // owner
- false, // autodelete
- false, // exclusive
- false,
- properties);
-
-
-
- if (lifetimePolicy == null || lifetimePolicy instanceof DeleteOnClose)
+ if(lifetimePolicy instanceof DeleteOnNoLinks)
{
- final Action<Connection_1_0> deleteQueueTask =
- new Action<Connection_1_0>()
- {
- public void performAction(Connection_1_0 session)
- {
- if (_vhost.getQueue(queueName) == tempQueue)
- {
- try
- {
- _vhost.removeQueue(tempQueue);
- }
- catch (QpidSecurityException e)
- {
- //TODO
- _logger.error("Error removing queue from vhost", e);
- }
- }
- }
- };
-
- _connection.addConnectionCloseTask(deleteQueueTask);
-
- queue.addQueueDeleteTask(new Action<AMQQueue>()
- {
- public void performAction(AMQQueue queue)
- {
- _connection.removeConnectionCloseTask(deleteQueueTask);
- }
-
-
- });
- }
- else if(lifetimePolicy instanceof DeleteOnNoLinks)
- {
-
- }
- else if(lifetimePolicy instanceof DeleteOnNoMessages)
- {
-
+ attributes.put(org.apache.qpid.server.model.Queue.LIFETIME_POLICY,
+ org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_NO_LINKS);
}
else if(lifetimePolicy instanceof DeleteOnNoLinksOrMessages)
{
-
+ attributes.put(org.apache.qpid.server.model.Queue.LIFETIME_POLICY,
+ org.apache.qpid.server.model.LifetimePolicy.IN_USE);
}
+ else if(lifetimePolicy instanceof DeleteOnClose)
+ {
+ attributes.put(org.apache.qpid.server.model.Queue.LIFETIME_POLICY,
+ org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_CONNECTION_CLOSE);
+ }
+ else if(lifetimePolicy instanceof DeleteOnNoMessages)
+ {
+ attributes.put(org.apache.qpid.server.model.Queue.LIFETIME_POLICY,
+ org.apache.qpid.server.model.LifetimePolicy.IN_USE);
+ }
+ else
+ {
+ attributes.put(org.apache.qpid.server.model.Queue.LIFETIME_POLICY,
+ org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_CONNECTION_CLOSE);
+ }
+
+
+ // TODO convert AMQP 1-0 node properties to queue attributes
+
+ final AMQQueue tempQueue = queue = _vhost.createQueue(this, attributes );
}
catch (QpidSecurityException e)
{
@@ -462,11 +450,6 @@
}
- public void forceEnd()
- {
- }
-
-
@Override
public UUID getId()
{
@@ -474,9 +457,9 @@
}
@Override
- public AMQConnectionModel getConnectionModel()
+ public Connection_1_0 getConnectionModel()
{
- return _connection.getModel();
+ return _connection;
}
@Override
@@ -489,14 +472,35 @@
@Override
public void close()
{
- // TODO - required for AMQSessionModel / management initiated closing
+ performCloseTasks();
+ _endpoint.end();
+ }
+
+ protected void performCloseTasks()
+ {
+
+ if(_closed.compareAndSet(false, true))
+ {
+ List<Action<? super Session_1_0>> taskList = new ArrayList<Action<? super Session_1_0>>(_taskList);
+ _taskList.clear();
+ for(Action<? super Session_1_0> task : taskList)
+ {
+ task.performAction(this);
+ }
+ }
}
@Override
public void close(AMQConstant cause, String message)
{
- // TODO - required for AMQSessionModel
+ performCloseTasks();
+ final End end = new End();
+ final Error theError = new Error();
+ theError.setDescription(message);
+ theError.setCondition(ConnectionError.CONNECTION_FORCED);
+ end.setError(theError);
+ _endpoint.end(end);
}
@Override
@@ -586,8 +590,7 @@
@Override
public int getChannelId()
{
- // TODO
- return 0;
+ return _endpoint.getSendingChannel();
}
@Override
@@ -609,13 +612,12 @@
connectionId,
getClientID(),
remoteAddress,
- _vhost.getName(), // TODO - virtual host
- 0) // TODO - channel)
- + "] ";
+ _vhost.getName(),
+ _endpoint.getSendingChannel()) + "] ";
}
@Override
- public int compareTo(AMQSessionModel o)
+ public int compareTo(Session_1_0 o)
{
return getId().compareTo(o.getId());
}
@@ -625,4 +627,18 @@
return _connection;
}
+ @Override
+ public void addDeleteTask(final Action<? super Session_1_0> task)
+ {
+ if(!_closed.get())
+ {
+ _taskList.add(task);
+ }
+ }
+
+ @Override
+ public void removeDeleteTask(final Action<? super Session_1_0> task)
+ {
+ _taskList.remove(task);
+ }
}
diff --git a/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java b/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
index 7d76a0e..3ded20ae 100644
--- a/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
+++ b/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
@@ -971,36 +971,18 @@
}
@Override
- public AuthorizationHolder getAuthorizationHolder()
- {
- return null;
- }
-
- @Override
- public void setAuthorizationHolder(final AuthorizationHolder principalHolder)
- {
-
- }
-
- @Override
- public void setExclusiveOwningSession(final AMQSessionModel owner)
- {
-
- }
-
- @Override
- public AMQSessionModel getExclusiveOwningSession()
- {
- return null;
- }
-
- @Override
public boolean isExclusive()
{
return false;
}
@Override
+ public boolean verifySessionAccess(final AMQSessionModel<?, ?> session)
+ {
+ return true;
+ }
+
+ @Override
public String getName()
{
return MANAGEMENT_NODE_NAME;
diff --git a/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/ExchangeMBean.java b/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/ExchangeMBean.java
index 407da0f..ed5e195 100644
--- a/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/ExchangeMBean.java
+++ b/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/ExchangeMBean.java
@@ -166,7 +166,7 @@
public boolean isAutoDelete()
{
- return _exchange.getLifetimePolicy() == LifetimePolicy.AUTO_DELETE;
+ return _exchange.getLifetimePolicy() != LifetimePolicy.PERMANENT;
}
public TabularData bindings() throws IOException, JMException
diff --git a/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java b/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
index 1365ceb..b44a752 100644
--- a/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
+++ b/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
@@ -50,6 +50,7 @@
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.ExclusivityPolicy;
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.QueueNotificationListener;
@@ -194,7 +195,7 @@
public boolean isAutoDelete()
{
- return _queue.getLifetimePolicy() == LifetimePolicy.AUTO_DELETE;
+ return _queue.getLifetimePolicy() != LifetimePolicy.PERMANENT;
}
public Long getMaximumMessageAge()
@@ -264,12 +265,29 @@
public boolean isExclusive()
{
- return (Boolean) _queue.getAttribute(Queue.EXCLUSIVE);
+ final Object attribute = _queue.getAttribute(Queue.EXCLUSIVE);
+ return attribute != null && attribute != ExclusivityPolicy.NONE;
}
public void setExclusive(boolean exclusive)
{
- _queue.setAttribute(Queue.EXCLUSIVE, isExclusive(), exclusive);
+ if(exclusive)
+ {
+ Object currentValue = _queue.getAttribute(Queue.EXCLUSIVE);
+ if(currentValue == null || currentValue == ExclusivityPolicy.NONE)
+ {
+ _queue.setAttribute(Queue.EXCLUSIVE, currentValue, ExclusivityPolicy.CONTAINER);
+ }
+ }
+ else
+ {
+ Object currentValue = _queue.getAttribute(Queue.EXCLUSIVE);
+ if(currentValue != null && currentValue != ExclusivityPolicy.NONE)
+ {
+ _queue.setAttribute(Queue.EXCLUSIVE, currentValue, ExclusivityPolicy.NONE);
+ }
+ }
+
}
public void setAlternateExchange(String exchangeName) throws OperationsException
diff --git a/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBean.java b/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBean.java
index 2c88f83..c39c3f7 100644
--- a/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBean.java
+++ b/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBean.java
@@ -181,8 +181,13 @@
throws IOException, JMException
{
final Map<String, Object> createArgs = processNewQueueArguments(queueName, owner, originalArguments);
- getConfiguredObject().createQueue(queueName, State.ACTIVE, durable, false, LifetimePolicy.PERMANENT, 0l,
- QueueArgumentsConverter.convertWireArgsToModel(createArgs));
+
+ final Map<String, Object> attributes = QueueArgumentsConverter.convertWireArgsToModel(createArgs);
+ attributes.put(Queue.NAME, queueName);
+ attributes.put(Queue.DURABLE, durable);
+ attributes.put(Queue.LIFETIME_POLICY, LifetimePolicy.PERMANENT);
+
+ getConfiguredObject().createQueue(attributes);
}
diff --git a/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/QueueMBeanTest.java b/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/QueueMBeanTest.java
index f94c206..2874168 100644
--- a/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/QueueMBeanTest.java
+++ b/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/QueueMBeanTest.java
@@ -40,6 +40,7 @@
import org.apache.qpid.server.jmx.mbeans.QueueMBean.GetMessageVisitor;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.ExclusivityPolicy;
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.Statistics;
@@ -154,7 +155,7 @@
public void testIsAutoDelete() throws Exception
{
- when(_mockQueue.getLifetimePolicy()).thenReturn(LifetimePolicy.AUTO_DELETE);
+ when(_mockQueue.getLifetimePolicy()).thenReturn(LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS);
assertTrue(_queueMBean.isAutoDelete());
}
@@ -224,22 +225,31 @@
testSetAttribute("flowResumeCapacity", Queue.QUEUE_FLOW_RESUME_SIZE_BYTES,1048576l , 2097152l);
}
+
+ /********** Other attributes **********/
+
+
public void testIsExclusive() throws Exception
{
- assertAttribute("exclusive", Boolean.TRUE, Queue.EXCLUSIVE);
+ when(_mockQueue.getAttribute(Queue.EXCLUSIVE)).thenReturn(ExclusivityPolicy.CONTAINER);
+ MBeanTestUtils.assertMBeanAttribute(_queueMBean, "exclusive", true);
}
public void testIsNotExclusive() throws Exception
{
- assertAttribute("exclusive", Boolean.FALSE, Queue.EXCLUSIVE);
+ when(_mockQueue.getAttribute(Queue.EXCLUSIVE)).thenReturn(ExclusivityPolicy.NONE);
+ MBeanTestUtils.assertMBeanAttribute(_queueMBean, "exclusive", false);
}
public void testSetExclusive() throws Exception
{
- testSetAttribute("exclusive", Queue.EXCLUSIVE, Boolean.FALSE , Boolean.TRUE);
- }
+ when(_mockQueue.getAttribute(Queue.EXCLUSIVE)).thenReturn(ExclusivityPolicy.NONE);
- /********** Other attributes **********/
+ MBeanTestUtils.setMBeanAttribute(_queueMBean, "exclusive", Boolean.TRUE);
+
+ verify(_mockQueue).setAttribute(Queue.EXCLUSIVE, ExclusivityPolicy.NONE, ExclusivityPolicy.CONTAINER);
+
+ }
public void testGetAlternateExchange()
{
diff --git a/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBeanTest.java b/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBeanTest.java
index 4240dd5..2dc2cb8 100644
--- a/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBeanTest.java
+++ b/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBeanTest.java
@@ -39,6 +39,7 @@
import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.queue.QueueArgumentsConverter;
+import org.mockito.ArgumentCaptor;
public class VirtualHostManagerMBeanTest extends TestCase
{
@@ -68,8 +69,15 @@
public void testCreateQueueWithNoOwner() throws Exception
{
_virtualHostManagerMBean.createNewQueue(TEST_QUEUE_NAME, null, true);
+ ArgumentCaptor<Map> argsCaptor = ArgumentCaptor.forClass(Map.class);
- verify(_mockVirtualHost).createQueue(TEST_QUEUE_NAME, State.ACTIVE, true, false, LifetimePolicy.PERMANENT, 0, EMPTY_ARGUMENT_MAP);
+ verify(_mockVirtualHost).createQueue(argsCaptor.capture());
+
+ Map actualAttributes = argsCaptor.getValue();
+ assertEquals(TEST_QUEUE_NAME, actualAttributes.get(Queue.NAME));
+ assertEquals(Boolean.TRUE,actualAttributes.get(Queue.DURABLE));
+ assertEquals(null,actualAttributes.get(Queue.OWNER));
+
}
/**
@@ -79,9 +87,15 @@
public void testCreateQueueWithOwnerMappedThroughToDescription() throws Exception
{
_virtualHostManagerMBean.createNewQueue(TEST_QUEUE_NAME, TEST_OWNER, true);
+ ArgumentCaptor<Map> argsCaptor = ArgumentCaptor.forClass(Map.class);
- Map<String, Object> expectedArguments = Collections.singletonMap(Queue.DESCRIPTION, (Object)TEST_OWNER);
- verify(_mockVirtualHost).createQueue(TEST_QUEUE_NAME, State.ACTIVE, true, false, LifetimePolicy.PERMANENT, 0, expectedArguments);
+ verify(_mockVirtualHost).createQueue(argsCaptor.capture());
+
+ Map actualAttributes = argsCaptor.getValue();
+ assertEquals(TEST_QUEUE_NAME,actualAttributes.get(Queue.NAME));
+ assertEquals(Boolean.TRUE,actualAttributes.get(Queue.DURABLE));
+ assertEquals(null,actualAttributes.get(Queue.OWNER));
+ assertEquals(TEST_OWNER, actualAttributes.get(Queue.DESCRIPTION));
}
public void testCreateQueueWithOwnerAndDescriptionDiscardsOwner() throws Exception
@@ -89,8 +103,15 @@
Map<String, Object> arguments = Collections.singletonMap(QueueArgumentsConverter.X_QPID_DESCRIPTION, (Object)TEST_DESCRIPTION);
_virtualHostManagerMBean.createNewQueue(TEST_QUEUE_NAME, TEST_OWNER, true, arguments);
- Map<String, Object> expectedArguments = Collections.singletonMap(Queue.DESCRIPTION, (Object)TEST_DESCRIPTION);
- verify(_mockVirtualHost).createQueue(TEST_QUEUE_NAME, State.ACTIVE, true, false, LifetimePolicy.PERMANENT, 0, expectedArguments);
+ ArgumentCaptor<Map> argsCaptor = ArgumentCaptor.forClass(Map.class);
+
+ verify(_mockVirtualHost).createQueue(argsCaptor.capture());
+
+ Map actualAttributes = argsCaptor.getValue();
+ assertEquals(TEST_QUEUE_NAME,actualAttributes.get(Queue.NAME));
+ assertEquals(Boolean.TRUE,actualAttributes.get(Queue.DURABLE));
+ assertEquals(null,actualAttributes.get(Queue.OWNER));
+ assertEquals(TEST_DESCRIPTION, actualAttributes.get(Queue.DESCRIPTION));
}
public void testDeleteQueue() throws Exception
diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/ModelTest.java b/java/systests/src/main/java/org/apache/qpid/server/queue/ModelTest.java
index 41d93b5..c6b2c9e 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/queue/ModelTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/queue/ModelTest.java
@@ -290,7 +290,7 @@
session.createQueue(new AMQShortString(queueName),
autoDelete, durable, exclusive);
- validateQueueViaJMX(queueName, exclusive ? connection.getClientID() : null, durable, autoDelete);
+ validateQueueViaJMX(queueName, (exclusive && durable &&!isBroker010()) ? connection.getClientID() : null, durable, autoDelete || (exclusive && !isBroker010() && !durable));
}
/**
diff --git a/java/systests/src/main/java/org/apache/qpid/server/store/MessageStoreTest.java b/java/systests/src/main/java/org/apache/qpid/server/store/MessageStoreTest.java
index bdcdbe2..709d8ae 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/store/MessageStoreTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/store/MessageStoreTest.java
@@ -23,7 +23,7 @@
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
+
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.log4j.Logger;
@@ -41,6 +41,9 @@
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.exchange.TopicExchange;
import org.apache.qpid.server.message.InstanceProperties;
+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.model.Queue;
import org.apache.qpid.server.protocol.v0_8.AMQMessage;
import org.apache.qpid.server.protocol.v0_8.MessageMetaData;
@@ -51,7 +54,6 @@
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.ConflationQueue;
import org.apache.qpid.server.queue.StandardQueue;
-import org.apache.qpid.server.security.QpidSecurityException;
import org.apache.qpid.server.txn.AutoCommitTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.BrokerTestHelper;
@@ -539,11 +541,10 @@
}
}
- private void setQueueExclusivity(boolean exclusive) throws AMQException
+ private void setQueueExclusivity(boolean exclusive) throws MessageSource.ExistingConsumerPreventsExclusive
{
AMQQueue queue = getVirtualHost().getQueue(durableExclusiveQueueName);
-
- queue.setExclusive(exclusive);
+ queue.setExclusivityPolicy(exclusive ? ExclusivityPolicy.CONTAINER : ExclusivityPolicy.NONE);
}
private void validateQueueExclusivityProperty(boolean expected)
@@ -587,7 +588,7 @@
assertEquals("Queue is not 'simple'", StandardQueue.class, queue.getClass());
}
- assertEquals("Queue owner is not as expected", queueOwner, queue.getOwner());
+ assertEquals("Queue owner is not as expected", exclusive ? queueOwner : null, queue.getOwner());
assertEquals("Queue durability is not as expected", durable, queue.isDurable());
assertEquals("Queue exclusivity is not as expected", exclusive, queue.isExclusive());
}
@@ -671,7 +672,7 @@
throws Exception
{
- Map<String,Object> queueArguments = null;
+ Map<String,Object> queueArguments = new HashMap<String, Object>();
if(usePriority || lastValueQueue)
{
@@ -680,19 +681,27 @@
if (usePriority)
{
- queueArguments = Collections.singletonMap(Queue.PRIORITIES, (Object) DEFAULT_PRIORTY_LEVEL);
+ queueArguments.put(Queue.PRIORITIES, DEFAULT_PRIORTY_LEVEL);
}
if (lastValueQueue)
{
- queueArguments = Collections.singletonMap(Queue.LVQ_KEY, (Object) LVQ_KEY);
+ queueArguments.put(Queue.LVQ_KEY, LVQ_KEY);
}
+ queueArguments.put(Queue.ID, UUIDGenerator.generateRandomUUID());
+ queueArguments.put(Queue.NAME, queueName);
+ queueArguments.put(Queue.DURABLE, durable);
+ queueArguments.put(Queue.LIFETIME_POLICY, LifetimePolicy.PERMANENT);
+ queueArguments.put(Queue.EXCLUSIVE, exclusive ? ExclusivityPolicy.CONTAINER : ExclusivityPolicy.NONE);
+ if(exclusive && queueOwner != null)
+ {
+ queueArguments.put(Queue.OWNER, queueOwner);
+ }
AMQQueue queue = null;
//Ideally we would be able to use the QueueDeclareHandler here.
- queue = getVirtualHost().createQueue(UUIDGenerator.generateRandomUUID(), queueName, durable, queueOwner, false, exclusive,
- false, queueArguments);
+ queue = getVirtualHost().createQueue(null, queueArguments);
validateQueueProperties(queue, usePriority, durable, exclusive, lastValueQueue);
diff --git a/java/systests/src/main/java/org/apache/qpid/systest/management/jmx/QueueManagementTest.java b/java/systests/src/main/java/org/apache/qpid/systest/management/jmx/QueueManagementTest.java
index 3c15a45..9319749 100644
--- a/java/systests/src/main/java/org/apache/qpid/systest/management/jmx/QueueManagementTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/systest/management/jmx/QueueManagementTest.java
@@ -130,9 +130,10 @@
public void testExclusiveQueueHasJmsClientIdAsOwner() throws Exception
{
- Queue tmpQueue = _session.createTemporaryQueue();
+ final String subName = "testOwner";
+ _session.createDurableSubscriber(getTestTopic(), subName);
- final String queueName = tmpQueue.getQueueName();
+ final String queueName = _connection.getClientID() + ":" + subName;
final ManagedQueue managedQueue = _jmxUtils.getManagedQueue(queueName);
assertNotNull(_connection.getClientID());
diff --git a/java/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java b/java/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java
index 9028130..1b27ee7 100644
--- a/java/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java
+++ b/java/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java
@@ -31,16 +31,7 @@
import javax.jms.JMSException;
import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.server.model.Binding;
-import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Connection;
-import org.apache.qpid.server.model.Exchange;
-import org.apache.qpid.server.model.LifetimePolicy;
-import org.apache.qpid.server.model.Port;
-import org.apache.qpid.server.model.Protocol;
-import org.apache.qpid.server.model.Queue;
-import org.apache.qpid.server.model.State;
-import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.model.*;
import org.apache.qpid.test.utils.TestBrokerConfiguration;
public class Asserts
@@ -100,7 +91,8 @@
assertEquals("Unexpected value of queue attribute " + Queue.QUEUE_TYPE, queueType, queueData.get(Queue.QUEUE_TYPE));
if (expectedAttributes == null)
{
- assertEquals("Unexpected value of queue attribute " + Queue.EXCLUSIVE, Boolean.FALSE, queueData.get(Queue.EXCLUSIVE));
+ assertEquals("Unexpected value of queue attribute " + Queue.EXCLUSIVE,
+ ExclusivityPolicy.NONE.name(), queueData.get(Queue.EXCLUSIVE));
assertEquals("Unexpected value of queue attribute " + Queue.MAXIMUM_DELIVERY_ATTEMPTS, 0,
queueData.get(Queue.MAXIMUM_DELIVERY_ATTEMPTS));
assertEquals("Unexpected value of queue attribute " + Queue.QUEUE_FLOW_CONTROL_SIZE_BYTES, 0,
diff --git a/java/systests/src/main/java/org/apache/qpid/systest/rest/QueueRestTest.java b/java/systests/src/main/java/org/apache/qpid/systest/rest/QueueRestTest.java
index fec516b..3833af3 100644
--- a/java/systests/src/main/java/org/apache/qpid/systest/rest/QueueRestTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/systest/rest/QueueRestTest.java
@@ -145,7 +145,6 @@
attributes.put(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_BYTES, 40000);
attributes.put(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES, 50000);
attributes.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, 10);
- attributes.put(Queue.EXCLUSIVE, true);
responseCode = getRestTestHelper().submitRequest("/rest/queue/test/" + queueName, "PUT", attributes);
assertEquals("Setting of queue attribites should be allowed", 200, responseCode);
@@ -158,7 +157,6 @@
assertEquals("Unexpected " + Queue.ALERT_THRESHOLD_MESSAGE_SIZE, 30000, queueData.get(Queue.ALERT_THRESHOLD_MESSAGE_SIZE) );
assertEquals("Unexpected " + Queue.ALERT_THRESHOLD_QUEUE_DEPTH_BYTES, 40000, queueData.get(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_BYTES) );
assertEquals("Unexpected " + Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES, 50000, queueData.get(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES) );
- assertEquals("Unexpected " + Queue.EXCLUSIVE, true, queueData.get(Queue.EXCLUSIVE) );
}
public void testPutCreateBinding() throws Exception
@@ -217,7 +215,7 @@
assertEquals("Unexpected binding attribute " + Consumer.NAME, "1", consumer.get(Consumer.NAME));
assertEquals("Unexpected binding attribute " + Consumer.DURABLE, Boolean.FALSE, consumer.get(Consumer.DURABLE));
- assertEquals("Unexpected binding attribute " + Consumer.LIFETIME_POLICY, LifetimePolicy.AUTO_DELETE.name(),
+ assertEquals("Unexpected binding attribute " + Consumer.LIFETIME_POLICY, LifetimePolicy.DELETE_ON_SESSION_END.name(),
consumer.get(Consumer.LIFETIME_POLICY));
assertEquals("Unexpected binding attribute " + Consumer.DISTRIBUTION_MODE, "MOVE",
consumer.get(Consumer.DISTRIBUTION_MODE));
diff --git a/java/systests/src/main/java/org/apache/qpid/systest/rest/VirtualHostRestTest.java b/java/systests/src/main/java/org/apache/qpid/systest/rest/VirtualHostRestTest.java
index 96ca0c2..d7b17fd 100644
--- a/java/systests/src/main/java/org/apache/qpid/systest/rest/VirtualHostRestTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/systest/rest/VirtualHostRestTest.java
@@ -36,7 +36,7 @@
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.VirtualHost;
-import org.apache.qpid.server.queue.AMQQueueFactory;
+import org.apache.qpid.server.queue.ConflationQueue;
import org.apache.qpid.server.virtualhost.StandardVirtualHostFactory;
import org.apache.qpid.test.utils.TestFileUtils;
import org.apache.qpid.util.FileUtils;
@@ -291,7 +291,7 @@
Asserts.assertQueue(queueName , "lvq", lvqQueue);
assertEquals("Unexpected value of queue attribute " + Queue.DURABLE, Boolean.TRUE, lvqQueue.get(Queue.DURABLE));
- assertEquals("Unexpected lvq key attribute", AMQQueueFactory.QPID_DEFAULT_LVQ_KEY, lvqQueue.get(Queue.LVQ_KEY));
+ assertEquals("Unexpected lvq key attribute", ConflationQueue.DEFAULT_LVQ_KEY, lvqQueue.get(Queue.LVQ_KEY));
}
public void testPutCreateSortedQueueWithoutKey() throws Exception
diff --git a/java/systests/src/main/java/org/apache/qpid/test/unit/client/MaxDeliveryCountTest.java b/java/systests/src/main/java/org/apache/qpid/test/unit/client/MaxDeliveryCountTest.java
index 6356b17..6473a77 100644
--- a/java/systests/src/main/java/org/apache/qpid/test/unit/client/MaxDeliveryCountTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/test/unit/client/MaxDeliveryCountTest.java
@@ -321,14 +321,7 @@
MessageConsumer consumer;
if(durableSub)
{
- if (isBroker010())
- {
- consumer = clientSession.createConsumer(clientSession.createQueue("clientid:" +getName() + AMQQueueFactory.DEFAULT_DLQ_NAME_SUFFIX));
- }
- else
- {
- consumer = clientSession.createDurableSubscriber(clientSession.createTopic(destName), getName() + AMQQueueFactory.DEFAULT_DLQ_NAME_SUFFIX);
- }
+ consumer = clientSession.createConsumer(clientSession.createQueue("clientid:" +getName() + AMQQueueFactory.DEFAULT_DLQ_NAME_SUFFIX));
}
else
{
diff --git a/java/test-profiles/Java010Excludes b/java/test-profiles/Java010Excludes
index 67c2fcd..cbdb0c1 100755
--- a/java/test-profiles/Java010Excludes
+++ b/java/test-profiles/Java010Excludes
@@ -35,6 +35,9 @@
org.apache.qpid.server.logging.ChannelLoggingTest#testChannelStartConsumerFlowStarted
org.apache.qpid.server.logging.ConsumerLoggingTest#testSubscriptionSuspend
org.apache.qpid.server.logging.ChannelLoggingTest#testChannelClosedOnQueueArgumentsMismatch
+// 0-10 exclusive queue is session exclusive and not container exclusive (so exclusive owner is not client-id)
+org.apache.qpid.server.logging.DurableQueueLoggingTest#testQueueCreateDurableExclusive
+org.apache.qpid.server.logging.TransientQueueLoggingTest#testQueueCreateDurableExclusive
// 0-10 is not supported by the MethodRegistry
org.apache.qpid.test.unit.close.JavaServerCloseRaceConditionTest#*
@@ -69,3 +72,6 @@
// Java 0-10 client does not support re-binding the queue to the same exchange
org.apache.qpid.server.queue.QueueBindTest#testQueueCanBeReboundOnTopicExchange
+
+// Java 0-10 exclusive queues are owned by sessions not by the client-id, so the owner is not meaningful from JMX
+org.apache.qpid.systest.management.jmx.QueueManagementTest#testExclusiveQueueHasJmsClientIdAsOwner
diff --git a/java/test-profiles/JavaPre010Excludes b/java/test-profiles/JavaPre010Excludes
index 7f9a20e..ff8bcfc 100644
--- a/java/test-profiles/JavaPre010Excludes
+++ b/java/test-profiles/JavaPre010Excludes
@@ -78,3 +78,8 @@
// QPID-3396
org.apache.qpid.test.unit.client.connection.ConnectionTest#testExceptionWhenUserPassIsRequired
+
+// Non durable exclusive queues are exclusive to a connection (and thus auto-delete - or at least
+// not permanent), also their owner is not the container (client-id) but the connection
+org.apache.qpid.server.logging.TransientQueueLoggingTest#testQueueCreateDurableExclusive
+
