You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by lq...@apache.org on 2017/06/06 08:47:39 UTC
[4/5] qpid-broker-j git commit: QPID-7625: [Java Broker] AMQP 1.0
Enforce LifetimePolicy DeleteOnClose and support TemporaryTopics
QPID-7625: [Java Broker] AMQP 1.0 Enforce LifetimePolicy DeleteOnClose and support TemporaryTopics
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/3bdc2e35
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/3bdc2e35
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/3bdc2e35
Branch: refs/heads/master
Commit: 3bdc2e35549cf0bd6bfe71f636787bdb25c7a39c
Parents: 5a872b4
Author: Lorenz Quack <lq...@apache.org>
Authored: Mon Jun 5 15:13:43 2017 +0100
Committer: Lorenz Quack <lq...@apache.org>
Committed: Tue Jun 6 09:13:03 2017 +0100
----------------------------------------------------------------------
.../qpid/server/exchange/AbstractExchange.java | 77 +++++++-
.../org/apache/qpid/server/model/Exchange.java | 6 +
.../qpid/server/model/LifetimePolicy.java | 1 +
.../org/apache/qpid/server/model/Queue.java | 5 +
.../apache/qpid/server/protocol/LinkModel.java | 4 +-
.../apache/qpid/server/queue/AbstractQueue.java | 89 +++++----
.../qpid/server/queue/CreatingLinkInfo.java | 34 ++++
.../qpid/server/queue/CreatingLinkInfoImpl.java | 55 ++++++
.../qpid/server/util/DeleteDeleteTask.java | 41 ++++
.../protocol/v1_0/LinkDefinitionImpl.java | 14 ++
.../qpid/server/protocol/v1_0/LinkImpl.java | 24 +++
.../protocol/v1_0/SendingLinkEndpoint.java | 24 ++-
.../qpid/server/protocol/v1_0/Session_1_0.java | 86 ++++----
.../v1_0/StandardReceivingLinkEndpoint.java | 16 +-
.../v1_0/type/messaging/DeleteOnClose.java | 20 +-
.../v1_0/type/messaging/DeleteOnNoLinks.java | 20 +-
.../messaging/DeleteOnNoLinksOrMessages.java | 20 +-
.../v1_0/type/messaging/DeleteOnNoMessages.java | 20 +-
.../protocol/v1_0/type/messaging/Source.java | 6 +-
.../protocol/v1_0/type/messaging/Target.java | 6 +-
.../protocol/v1_0/store/LinkStoreTestCase.java | 5 +-
.../qpid/tests/protocol/v1_0/BrokerAdmin.java | 6 +
.../v1_0/EmbeddedBrokerPerClassAdminImpl.java | 28 +++
.../v1_0/ExternalQpidBrokerAdminImpl.java | 13 ++
.../tests/protocol/v1_0/FrameTransport.java | 6 +-
.../apache/qpid/tests/protocol/v1_0/Utils.java | 81 ++++++++
.../bindmapjms/TemporaryDestinationTest.java | 133 +++++++++++++
.../v1_0/messaging/DeleteOnCloseTest.java | 196 +++++++++++++++++++
.../org/apache/qpid/systest/rest/Asserts.java | 6 +-
.../temporaryqueue/TemporaryQueueTest.java | 9 -
.../test/unit/topic/TemporaryTopicTest.java | 8 +-
test-profiles/Java10BrokenTestsExcludes | 9 +-
32 files changed, 889 insertions(+), 179 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java b/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
index f06fcdf..92c5bba 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
@@ -21,6 +21,7 @@
package org.apache.qpid.server.exchange;
import java.security.AccessControlException;
+import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -36,13 +37,15 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import javax.security.auth.Subject;
+
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.server.exchange.ExchangeDefaults;
import org.apache.qpid.server.binding.BindingImpl;
+import org.apache.qpid.server.configuration.IllegalConfigurationException;
import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.logging.messages.BindingMessages;
@@ -66,9 +69,14 @@ import org.apache.qpid.server.model.PublishingLink;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.StateTransition;
+import org.apache.qpid.server.protocol.LinkModel;
+import org.apache.qpid.server.queue.CreatingLinkInfo;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.security.access.Operation;
import org.apache.qpid.server.store.StorableMessageMetaData;
+import org.apache.qpid.server.util.Action;
+import org.apache.qpid.server.util.Deletable;
+import org.apache.qpid.server.util.DeleteDeleteTask;
import org.apache.qpid.server.util.FixedKeyMapCreator;
import org.apache.qpid.server.virtualhost.ExchangeIsAlternateException;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
@@ -96,6 +104,8 @@ public abstract class AbstractExchange<T extends AbstractExchange<T>>
private Exchange<?> _alternateExchange;
@ManagedAttributeField
private UnroutableMessageBehaviour _unroutableMessageBehaviour;
+ @ManagedAttributeField
+ private CreatingLinkInfo _creatingLinkInfo;
private QueueManagingVirtualHost<?> _virtualHost;
@@ -118,6 +128,7 @@ public abstract class AbstractExchange<T extends AbstractExchange<T>>
private final List<Binding> _bindings = new CopyOnWriteArrayList<>();
private final ConcurrentMap<MessageSender, Integer> _linkedSenders = new ConcurrentHashMap<>();
+ private final List<Action<? super Deletable<?>>> _deleteTaskList = new CopyOnWriteArrayList<>();
public AbstractExchange(Map<String, Object> attributes, QueueManagingVirtualHost<?> vhost)
{
@@ -153,6 +164,15 @@ public abstract class AbstractExchange<T extends AbstractExchange<T>>
|| name.startsWith("amq.") || name.startsWith("qpid.");
}
+ @Override
+ protected void validateOnCreate()
+ {
+ super.validateOnCreate();
+ if (getCreatingLinkInfo() != null && !isSystemProcess())
+ {
+ throw new IllegalConfigurationException(String.format("Cannot specify creatingLinkInfo for exchange '%s'", getName()));
+ }
+ }
@Override
protected void onOpen()
@@ -175,7 +195,30 @@ public abstract class AbstractExchange<T extends AbstractExchange<T>>
}
}
}
- // Log Exchange creation
+
+ if (getLifetimePolicy() == LifetimePolicy.DELETE_ON_CREATING_LINK_CLOSE)
+ {
+ if (_creatingLinkInfo != null)
+ {
+ final LinkModel link;
+ if (_creatingLinkInfo.isSendingLink())
+ {
+ link = _virtualHost.getSendingLink(_creatingLinkInfo.getRemoteContainerId(), _creatingLinkInfo.getLinkName());
+ }
+ else
+ {
+ link = _virtualHost.getReceivingLink(_creatingLinkInfo.getRemoteContainerId(), _creatingLinkInfo.getLinkName());
+ }
+ addLifetimeConstraint(link);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Exchanges created with a lifetime policy of "
+ + getLifetimePolicy()
+ + " must be created from a AMQP 1.0 link.");
+ }
+ }
+
getEventLogger().message(ExchangeMessages.CREATED(getType(), getName(), isDurable()));
}
@@ -185,11 +228,34 @@ public abstract class AbstractExchange<T extends AbstractExchange<T>>
return _virtualHost.getEventLogger();
}
+ private void performDeleteTasks()
+ {
+ for (Action<? super Deletable<?>> task : _deleteTaskList)
+ {
+ task.performAction(null);
+ }
+
+ _deleteTaskList.clear();
+ }
+
public boolean isAutoDelete()
{
return getLifetimePolicy() != LifetimePolicy.PERMANENT;
}
+ private void addLifetimeConstraint(final Deletable<? extends Deletable> lifetimeObject)
+ {
+ final Action<Deletable> deleteExchangeTask = object -> Subject.doAs(getSubjectWithAddedSystemRights(),
+ (PrivilegedAction<Void>) () ->
+ {
+ AbstractExchange.this.delete();
+ return null;
+ });
+
+ lifetimeObject.addDeleteTask(deleteExchangeTask);
+ _deleteTaskList.add(new DeleteDeleteTask(lifetimeObject, deleteExchangeTask));
+ }
+
private void deleteWithChecks()
{
if(hasReferrers())
@@ -204,6 +270,7 @@ public abstract class AbstractExchange<T extends AbstractExchange<T>>
if(_closed.compareAndSet(false,true))
{
+ performDeleteTasks();
for(Binding b : _bindings)
{
@@ -641,6 +708,12 @@ public abstract class AbstractExchange<T extends AbstractExchange<T>>
return durableBindings;
}
+ @Override
+ public CreatingLinkInfo getCreatingLinkInfo()
+ {
+ return _creatingLinkInfo;
+ }
+
private MessageDestination getAttainedMessageDestination(final String name)
{
MessageDestination destination = getVirtualHost().getAttainedQueue(name);
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java b/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java
index 120ce05..1f30ec1 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java
@@ -27,6 +27,8 @@ import org.apache.qpid.server.exchange.ExchangeReferrer;
import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageSender;
+import org.apache.qpid.server.queue.CreatingLinkInfo;
+import org.apache.qpid.server.util.Deletable;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
@ManagedObject( description = Exchange.CLASS_DESCRIPTION,
@@ -43,6 +45,7 @@ public interface Exchange<X extends Exchange<X>> extends ConfiguredObject<X>, Me
String ALTERNATE_EXCHANGE = "alternateExchange";
String DURABLE_BINDINGS = "durableBindings";
String UNROUTABLE_MESSAGE_BEHAVIOUR = "unroutableMessageBehaviour";
+ String CREATING_LINK_INFO = "creatingLinkInfo";
enum UnroutableMessageBehaviour
{
@@ -66,6 +69,9 @@ public interface Exchange<X extends Exchange<X>> extends ConfiguredObject<X>, Me
@DerivedAttribute(persist = true)
Collection<Binding> getDurableBindings();
+ @ManagedAttribute(immutable = true, description = "Information about the AMQP 1.0 Link that created this Exchange if any.")
+ CreatingLinkInfo getCreatingLinkInfo();
+
// Statistics
@ManagedStatistic(statisticType = StatisticType.POINT_IN_TIME, units = StatisticUnit.COUNT, label = "Bindings")
long getBindingCount();
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-core/src/main/java/org/apache/qpid/server/model/LifetimePolicy.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/model/LifetimePolicy.java b/broker-core/src/main/java/org/apache/qpid/server/model/LifetimePolicy.java
index 1e45f8f..8bf9642 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/LifetimePolicy.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/LifetimePolicy.java
@@ -27,5 +27,6 @@ public enum LifetimePolicy
DELETE_ON_SESSION_END,
DELETE_ON_NO_OUTBOUND_LINKS,
DELETE_ON_NO_LINKS,
+ DELETE_ON_CREATING_LINK_CLOSE,
IN_USE
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java b/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java
index 5e0ca02..aefbc87 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java
@@ -34,6 +34,7 @@ import org.apache.qpid.server.message.MessageInfo;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.queue.CreatingLinkInfo;
import org.apache.qpid.server.queue.NotificationCheck;
import org.apache.qpid.server.queue.QueueConsumer;
import org.apache.qpid.server.queue.QueueEntry;
@@ -73,6 +74,7 @@ public interface Queue<X extends Queue<X>> extends ConfiguredObject<X>,
String MAXIMUM_DELIVERY_ATTEMPTS = "maximumDeliveryAttempts";
String NO_LOCAL = "noLocal";
String OWNER = "owner";
+ String CREATING_LINK_INFO = "creatingLinkInfo";
String QUEUE_FLOW_STOPPED = "queueFlowStopped";
String MAXIMUM_MESSAGE_TTL = "maximumMessageTtl";
@@ -126,6 +128,9 @@ public interface Queue<X extends Queue<X>> extends ConfiguredObject<X>,
@DerivedAttribute( persist = true )
String getOwner();
+ @ManagedAttribute(immutable = true, description = "Information about the AMQP 1.0 Link that created this Queue if any.")
+ CreatingLinkInfo getCreatingLinkInfo();
+
@SuppressWarnings("unused")
@ManagedAttribute
boolean isNoLocal();
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-core/src/main/java/org/apache/qpid/server/protocol/LinkModel.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/protocol/LinkModel.java b/broker-core/src/main/java/org/apache/qpid/server/protocol/LinkModel.java
index 16120a3..970a733 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/protocol/LinkModel.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/protocol/LinkModel.java
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.server.protocol;
-public interface LinkModel
+import org.apache.qpid.server.util.Deletable;
+
+public interface LinkModel extends Deletable<LinkModel>
{
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
index 47f7315..129f9d2 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
@@ -98,6 +98,7 @@ import org.apache.qpid.server.model.preferences.GenericPrincipal;
import org.apache.qpid.server.plugin.MessageConverter;
import org.apache.qpid.server.plugin.MessageFilterFactory;
import org.apache.qpid.server.plugin.QpidServiceLoader;
+import org.apache.qpid.server.protocol.LinkModel;
import org.apache.qpid.server.protocol.MessageConverterRegistry;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.security.access.Operation;
@@ -114,6 +115,7 @@ 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.DeleteDeleteTask;
import org.apache.qpid.server.util.MapValueConverter;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.server.virtualhost.HouseKeepingTask;
@@ -197,9 +199,7 @@ public abstract class AbstractQueue<X extends AbstractQueue<X>>
private final AtomicBoolean _deleted = new AtomicBoolean(false);
private final SettableFuture<Integer> _deleteFuture = SettableFuture.create();
- private final List<Action<? super X>> _deleteTaskList =
- new CopyOnWriteArrayList<>();
-
+ private final List<Action<? super X>> _deleteTaskList = new CopyOnWriteArrayList<>();
private LogSubject _logSubject;
@@ -244,6 +244,8 @@ public abstract class AbstractQueue<X extends AbstractQueue<X>>
private long _maximumQueueDepthMessages;
@ManagedAttributeField
private long _maximumQueueDepthBytes;
+ @ManagedAttributeField
+ private CreatingLinkInfo _creatingLinkInfo;
private static final int RECOVERING = 1;
private static final int COMPLETING_RECOVERY = 2;
@@ -316,6 +318,16 @@ public abstract class AbstractQueue<X extends AbstractQueue<X>>
}
@Override
+ protected void validateOnCreate()
+ {
+ super.validateOnCreate();
+ if (getCreatingLinkInfo() != null && !isSystemProcess())
+ {
+ throw new IllegalConfigurationException(String.format("Cannot specify creatingLinkInfo for queue '%s'", getName()));
+ }
+ }
+
+ @Override
public void onValidate()
{
super.onValidate();
@@ -441,6 +453,28 @@ public abstract class AbstractQueue<X extends AbstractQueue<X>>
+ " must be created from a connection.");
}
}
+ else if (getLifetimePolicy() == LifetimePolicy.DELETE_ON_CREATING_LINK_CLOSE)
+ {
+ if (_creatingLinkInfo != null)
+ {
+ final LinkModel link;
+ if (_creatingLinkInfo.isSendingLink())
+ {
+ link = _virtualHost.getSendingLink(_creatingLinkInfo.getRemoteContainerId(), _creatingLinkInfo.getLinkName());
+ }
+ else
+ {
+ link = _virtualHost.getReceivingLink(_creatingLinkInfo.getRemoteContainerId(), _creatingLinkInfo.getLinkName());
+ }
+ addLifetimeConstraint(link);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Queues created with a lifetime policy of "
+ + getLifetimePolicy()
+ + " must be created from a AMQP 1.0 link.");
+ }
+ }
// Log the creation of this Queue.
@@ -540,23 +574,12 @@ public abstract class AbstractQueue<X extends AbstractQueue<X>>
private void addLifetimeConstraint(final Deletable<? extends Deletable> lifetimeObject)
{
- final Action<Deletable> deleteQueueTask = new Action<Deletable>()
- {
- @Override
- public void performAction(final Deletable object)
- {
- Subject.doAs(getSubjectWithAddedSystemRights(),
- new PrivilegedAction<Void>()
- {
- @Override
- public Void run()
- {
- AbstractQueue.this.delete();
- return null;
- }
- });
- }
- };
+ final Action<Deletable> deleteQueueTask = object -> Subject.doAs(getSubjectWithAddedSystemRights(),
+ (PrivilegedAction<Void>) () ->
+ {
+ AbstractQueue.this.delete();
+ return null;
+ });
lifetimeObject.addDeleteTask(deleteQueueTask);
addDeleteTask(new DeleteDeleteTask(lifetimeObject, deleteQueueTask));
@@ -695,6 +718,12 @@ public abstract class AbstractQueue<X extends AbstractQueue<X>>
return null;
}
+ @Override
+ public CreatingLinkInfo getCreatingLinkInfo()
+ {
+ return _creatingLinkInfo;
+ }
+
public VirtualHost<?> getVirtualHost()
{
return _virtualHost;
@@ -2204,26 +2233,6 @@ public abstract class AbstractQueue<X extends AbstractQueue<X>>
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);
- }
- }
-
abstract class BaseMessageContent implements Content, CustomRestHeaders
{
public static final int UNLIMITED = -1;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-core/src/main/java/org/apache/qpid/server/queue/CreatingLinkInfo.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/CreatingLinkInfo.java b/broker-core/src/main/java/org/apache/qpid/server/queue/CreatingLinkInfo.java
new file mode 100644
index 0000000..21832c4
--- /dev/null
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/CreatingLinkInfo.java
@@ -0,0 +1,34 @@
+/*
+ * 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.queue;
+
+import org.apache.qpid.server.model.ManagedAttributeValue;
+import org.apache.qpid.server.model.ManagedAttributeValueType;
+
+@ManagedAttributeValueType
+public interface CreatingLinkInfo extends ManagedAttributeValue
+{
+ boolean isSendingLink();
+
+ String getRemoteContainerId();
+
+ String getLinkName();
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-core/src/main/java/org/apache/qpid/server/queue/CreatingLinkInfoImpl.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/CreatingLinkInfoImpl.java b/broker-core/src/main/java/org/apache/qpid/server/queue/CreatingLinkInfoImpl.java
new file mode 100644
index 0000000..c4edb3f
--- /dev/null
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/CreatingLinkInfoImpl.java
@@ -0,0 +1,55 @@
+/*
+ * 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.queue;
+
+public class CreatingLinkInfoImpl implements CreatingLinkInfo
+{
+ private final boolean _isSendingLink;
+ private final String _remoteContainerId;
+ private final String _linkName;
+
+ public CreatingLinkInfoImpl(final boolean isSendingLink,
+ final String remoteContainerId,
+ final String linkName)
+ {
+ _isSendingLink = isSendingLink;
+ _remoteContainerId = remoteContainerId;
+ _linkName = linkName;
+ }
+
+ @Override
+ public boolean isSendingLink()
+ {
+ return _isSendingLink;
+ }
+
+ @Override
+ public String getRemoteContainerId()
+ {
+ return _remoteContainerId;
+ }
+
+ @Override
+ public String getLinkName()
+ {
+ return _linkName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-core/src/main/java/org/apache/qpid/server/util/DeleteDeleteTask.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/util/DeleteDeleteTask.java b/broker-core/src/main/java/org/apache/qpid/server/util/DeleteDeleteTask.java
new file mode 100644
index 0000000..f95e7a0
--- /dev/null
+++ b/broker-core/src/main/java/org/apache/qpid/server/util/DeleteDeleteTask.java
@@ -0,0 +1,41 @@
+/*
+ * 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 class DeleteDeleteTask implements Action<Deletable>
+{
+
+ private final Deletable<? extends Deletable> _lifetimeObject;
+ private final Action<? super Deletable> _deleteTask;
+
+ public DeleteDeleteTask(final Deletable<? extends Deletable> lifetimeObject,
+ final Action<? super Deletable> deleteTask)
+ {
+ _lifetimeObject = lifetimeObject;
+ _deleteTask = deleteTask;
+ }
+
+ @Override
+ public void performAction(final Deletable object)
+ {
+ _lifetimeObject.removeDeleteTask(_deleteTask);
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinitionImpl.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinitionImpl.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinitionImpl.java
index d6fc95f..07dabf5 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinitionImpl.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkDefinitionImpl.java
@@ -20,9 +20,11 @@
package org.apache.qpid.server.protocol.v1_0;
+import org.apache.qpid.server.protocol.LinkModel;
import org.apache.qpid.server.protocol.v1_0.type.BaseSource;
import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
+import org.apache.qpid.server.util.Action;
public class LinkDefinitionImpl<S extends BaseSource, T extends BaseTarget> implements LinkDefinition<S, T>
{
@@ -74,4 +76,16 @@ public class LinkDefinitionImpl<S extends BaseSource, T extends BaseTarget> impl
{
return _target;
}
+
+ @Override
+ public void addDeleteTask(final Action<? super LinkModel> task)
+ {
+ throw new UnsupportedOperationException("");
+ }
+
+ @Override
+ public void removeDeleteTask(final Action<? super LinkModel> task)
+ {
+ throw new UnsupportedOperationException("");
+ }
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkImpl.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkImpl.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkImpl.java
index 77c5327..78d0bd2 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkImpl.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/LinkImpl.java
@@ -19,8 +19,10 @@
package org.apache.qpid.server.protocol.v1_0;
+import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import com.google.common.util.concurrent.FutureCallback;
@@ -32,6 +34,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.server.model.AbstractConfiguredObject;
+import org.apache.qpid.server.protocol.LinkModel;
import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
import org.apache.qpid.server.protocol.v1_0.type.BaseSource;
import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
@@ -44,6 +47,7 @@ import org.apache.qpid.server.protocol.v1_0.type.transport.Attach;
import org.apache.qpid.server.protocol.v1_0.type.transport.Error;
import org.apache.qpid.server.protocol.v1_0.type.transport.LinkError;
import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
+import org.apache.qpid.server.util.Action;
public class LinkImpl<S extends BaseSource, T extends BaseTarget> implements Link_1_0<S, T>
{
@@ -59,6 +63,7 @@ public class LinkImpl<S extends BaseSource, T extends BaseTarget> implements Lin
private volatile S _source;
private volatile T _target;
private boolean _stealingInProgress;
+ private final Queue<Action<? super Link_1_0<S, T>>> _deleteTasks = new ConcurrentLinkedQueue<>();
public LinkImpl(final String remoteContainerId, final String linkName, final Role role, final LinkRegistry linkRegistry)
{
@@ -113,6 +118,13 @@ public class LinkImpl<S extends BaseSource, T extends BaseTarget> implements Lin
@Override
public void linkClosed()
{
+ Iterator<Action<? super Link_1_0<S, T>>> iterator = _deleteTasks.iterator();
+ while (iterator.hasNext())
+ {
+ final Action<? super Link_1_0<S, T>> deleteTask = iterator.next();
+ deleteTask.performAction(this);
+ iterator.remove();
+ }
discardEndpoint();
_linkRegistry.linkClosed(this);
}
@@ -299,6 +311,18 @@ public class LinkImpl<S extends BaseSource, T extends BaseTarget> implements Lin
}
}
+ @Override
+ public void addDeleteTask(final Action<? super LinkModel> task)
+ {
+ _deleteTasks.add(task);
+ }
+
+ @Override
+ public void removeDeleteTask(final Action<? super LinkModel> task)
+ {
+ _deleteTasks.remove(task);
+ }
+
private class ThiefInformation
{
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLinkEndpoint.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLinkEndpoint.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLinkEndpoint.java
index dc1ea8c..b9f1e7b 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLinkEndpoint.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLinkEndpoint.java
@@ -48,13 +48,13 @@ import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.NotFoundException;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
-import org.apache.qpid.server.protocol.v1_0.type.BaseTarget;
import org.apache.qpid.server.protocol.v1_0.type.Binary;
import org.apache.qpid.server.protocol.v1_0.type.DeliveryState;
import org.apache.qpid.server.protocol.v1_0.type.Outcome;
import org.apache.qpid.server.protocol.v1_0.type.Symbol;
import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
import org.apache.qpid.server.protocol.v1_0.type.messaging.Accepted;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.DeleteOnNoMessages;
import org.apache.qpid.server.protocol.v1_0.type.messaging.Filter;
import org.apache.qpid.server.protocol.v1_0.type.messaging.Modified;
import org.apache.qpid.server.protocol.v1_0.type.messaging.NoLocalFilter;
@@ -292,13 +292,13 @@ public class SendingLinkEndpoint extends AbstractLinkEndpoint<Source, Target>
Source newSource = (Source) attach.getSource();
Source oldSource = getSource();
- final SendingDestination destination = getSession().getSendingDestination(getLinkName(), oldSource);
+ final SendingDestination destination = getSession().getSendingDestination(getLink(), oldSource);
prepareConsumerOptionsAndFilters(destination);
if (getDestination() instanceof ExchangeDestination && !Boolean.TRUE.equals(newSource.getDynamic()))
{
final SendingDestination newDestination =
- getSession().getSendingDestination(getLinkName(), newSource);
+ getSession().getSendingDestination(getLink(), newSource);
if (getSession().updateSourceForSubscription(this, newSource, newDestination))
{
setDestination(newDestination);
@@ -323,13 +323,13 @@ public class SendingLinkEndpoint extends AbstractLinkEndpoint<Source, Target>
Source newSource = (Source) attach.getSource();
Source oldSource = getSource();
- final SendingDestination destination = getSession().getSendingDestination(getLinkName(), oldSource);
+ final SendingDestination destination = getSession().getSendingDestination(getLink(), oldSource);
prepareConsumerOptionsAndFilters(destination);
if (getDestination() instanceof ExchangeDestination && !Boolean.TRUE.equals(newSource.getDynamic()))
{
final SendingDestination newDestination =
- getSession().getSendingDestination(getLinkName(), newSource);
+ getSession().getSendingDestination(getLink(), newSource);
if (getSession().updateSourceForSubscription(this, newSource, newDestination))
{
setDestination(newDestination);
@@ -359,7 +359,7 @@ public class SendingLinkEndpoint extends AbstractLinkEndpoint<Source, Target>
throw new AmqpErrorException(new Error(AmqpError.NOT_FOUND, ""));
}
- final SendingDestination destination = getSession().getSendingDestination(getLinkName(), getSource());
+ final SendingDestination destination = getSession().getSendingDestination(getLink(), getSource());
prepareConsumerOptionsAndFilters(destination);
attachReceived(attach);
@@ -639,13 +639,23 @@ public class SendingLinkEndpoint extends AbstractLinkEndpoint<Source, Target>
source.setAddress(attachSource.getAddress());
source.setDynamic(attachSource.getDynamic());
+ if (Boolean.TRUE.equals(attachSource.getDynamic()) && attachSource.getDynamicNodeProperties() != null)
+ {
+ Map<Symbol, Object> dynamicNodeProperties = new HashMap<>();
+ if (attachSource.getDynamicNodeProperties().containsKey(Session_1_0.LIFETIME_POLICY))
+ {
+ dynamicNodeProperties.put(Session_1_0.LIFETIME_POLICY,
+ attachSource.getDynamicNodeProperties().get(Session_1_0.LIFETIME_POLICY));
+ }
+ source.setDynamicNodeProperties(dynamicNodeProperties);
+ }
source.setDurable(TerminusDurability.min(attachSource.getDurable(),
getLink().getHighestSupportedTerminusDurability()));
source.setExpiryPolicy(attachSource.getExpiryPolicy());
source.setDistributionMode(attachSource.getDistributionMode());
source.setFilter(attachSource.getFilter());
source.setCapabilities(attachSource.getCapabilities());
- final SendingDestination destination = getSession().getSendingDestination(attach.getName(), source);
+ final SendingDestination destination = getSession().getSendingDestination(getLink(), source);
source.setCapabilities(destination.getCapabilities());
if (destination instanceof ExchangeDestination)
{
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
index 6183b8f..3ec200b 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
@@ -46,6 +46,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import javax.security.auth.Subject;
+import com.google.common.collect.Sets;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
@@ -64,6 +65,7 @@ import org.apache.qpid.server.logging.messages.ChannelMessages;
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.model.AbstractConfiguredObject;
+import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.Consumer;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.ExclusivityPolicy;
@@ -108,6 +110,8 @@ import org.apache.qpid.server.protocol.v1_0.type.transport.LinkError;
import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
import org.apache.qpid.server.protocol.v1_0.type.transport.SessionError;
import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer;
+import org.apache.qpid.server.queue.CreatingLinkInfo;
+import org.apache.qpid.server.queue.CreatingLinkInfoImpl;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.session.AbstractAMQPSession;
import org.apache.qpid.server.transport.AMQPConnection;
@@ -121,7 +125,7 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
{
public static final Symbol DELAYED_DELIVERY = Symbol.valueOf("DELAYED_DELIVERY");
private static final Logger _logger = LoggerFactory.getLogger(Session_1_0.class);
- private static final Symbol LIFETIME_POLICY = Symbol.valueOf("lifetime-policy");
+ public static final Symbol LIFETIME_POLICY = Symbol.valueOf("lifetime-policy");
private static final EnumSet<SessionState> END_STATES =
EnumSet.of(SessionState.END_RECVD, SessionState.END_PIPE, SessionState.END_SENT, SessionState.ENDED);
@@ -683,15 +687,17 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
return _accessControllerContext;
}
- public ReceivingDestination getReceivingDestination(final Target target) throws AmqpErrorException
+ public ReceivingDestination getReceivingDestination(final Link_1_0<?, ?> link,
+ final Target target) throws AmqpErrorException
{
final ReceivingDestination destination;
if (target != null)
{
if (Boolean.TRUE.equals(target.getDynamic()))
{
- MessageDestination tempQueue = createDynamicDestination(target.getDynamicNodeProperties());
- target.setAddress(tempQueue.getName());
+ MessageDestination tempDestination = createDynamicDestination(link, target.getDynamicNodeProperties(), target.getCapabilities());
+ // TODO: avoid NPE
+ target.setAddress(tempDestination.getName());
}
String addr = target.getAddress();
@@ -793,13 +799,14 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
return false;
}
- public SendingDestination getSendingDestination(final String linkName, final Source source) throws AmqpErrorException
+ public SendingDestination getSendingDestination(final Link_1_0<?, ?> link,
+ final Source source) throws AmqpErrorException
{
SendingDestination destination = null;
if (Boolean.TRUE.equals(source.getDynamic()))
{
- MessageSource tempQueue = createDynamicSource(source.getDynamicNodeProperties());
+ MessageSource tempQueue = createDynamicSource(link, source.getDynamicNodeProperties());
source.setAddress(tempQueue.getName()); // todo : temporary topic
}
@@ -808,7 +815,7 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
{
if (!address.startsWith("/") && address.contains("/"))
{
- destination = createExchangeDestination(address, linkName, source);
+ destination = createExchangeDestination(address, link.getName(), source);
}
else
{
@@ -819,7 +826,7 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
}
else
{
- destination = createExchangeDestination(address, null, linkName, source);
+ destination = createExchangeDestination(address, null, link.getName(), source);
}
}
}
@@ -1023,17 +1030,16 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
return false;
}
- private MessageSource createDynamicSource(Map properties)
+ private MessageSource createDynamicSource(final Link_1_0<?, ?> link,
+ Map properties)
{
final String queueName = _primaryDomain + "TempQueue" + UUID.randomUUID().toString();
- MessageSource queue = null;
try
{
- Map<String, Object> attributes = convertDynamicNodePropertiesToAttributes(properties, queueName);
+ Map<String, Object> attributes = convertDynamicNodePropertiesToAttributes(link, properties, queueName);
-
-
- queue = getAddressSpace().createMessageSource(MessageSource.class, attributes);
+ return Subject.doAs(getSubjectWithAddedSystemRights(),
+ (PrivilegedAction<MessageSource>) () -> getAddressSpace().createMessageSource(MessageSource.class, attributes));
}
catch (AccessControlException e)
{
@@ -1042,28 +1048,36 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
error.setDescription(e.getMessage());
_connection.close(error);
+ return null;
}
catch (AbstractConfiguredObject.DuplicateNameException e)
{
_logger.error("A temporary queue was created with a name which collided with an existing queue name");
throw new ConnectionScopedRuntimeException(e);
}
-
- return queue;
}
- private MessageDestination createDynamicDestination(Map properties)
+ private MessageDestination createDynamicDestination(final Link_1_0<?, ?> link,
+ Map properties,
+ final Symbol[] capabilities)
{
- final String queueName = _primaryDomain + "TempQueue" + UUID.randomUUID().toString();
- MessageDestination queue = null;
+ final Set<Symbol> capabilitySet = capabilities == null ? Collections.emptySet() : Sets.newHashSet(capabilities);
+ boolean isTopic = capabilitySet.contains(Symbol.valueOf("temporary-topic")) || capabilitySet.contains(Symbol.valueOf("topic"));
+ final String destName = _primaryDomain + (isTopic ? "TempTopic" : "TempQueue") + UUID.randomUUID().toString();
try
{
- Map<String, Object> attributes = convertDynamicNodePropertiesToAttributes(properties, queueName);
+ Map<String, Object> attributes = convertDynamicNodePropertiesToAttributes(link, properties, destName);
+ Class<? extends MessageDestination> clazz = isTopic ? Exchange.class : MessageDestination.class;
+ if (isTopic)
+ {
+ attributes.put(Exchange.TYPE, ExchangeDefaults.FANOUT_EXCHANGE_CLASS);
+ }
- queue = getAddressSpace().createMessageDestination(MessageDestination.class, attributes);
+ return Subject.doAs(getSubjectWithAddedSystemRights(),
+ (PrivilegedAction<MessageDestination>) () -> getAddressSpace().createMessageDestination(clazz, attributes));
}
catch (AccessControlException e)
{
@@ -1072,50 +1086,54 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
error.setDescription(e.getMessage());
_connection.close(error);
+ return null;
}
catch (AbstractConfiguredObject.DuplicateNameException e)
{
- _logger.error("A temporary queue was created with a name which collided with an existing queue name");
+ _logger.error("A temporary destination was created with a name which collided with an existing destination name '{}'", destName);
throw new ConnectionScopedRuntimeException(e);
}
-
- return queue;
}
- private Map<String, Object> convertDynamicNodePropertiesToAttributes(final Map properties, final String queueName)
+ private Map<String, Object> convertDynamicNodePropertiesToAttributes(final Link_1_0<?, ?> link,
+ final Map properties,
+ final String nodeName)
{
// TODO convert AMQP 1-0 node properties to queue attributes
LifetimePolicy lifetimePolicy = properties == null
? null
: (LifetimePolicy) properties.get(LIFETIME_POLICY);
+
Map<String,Object> attributes = new HashMap<>();
- attributes.put(Queue.ID, UUID.randomUUID());
- attributes.put(Queue.NAME, queueName);
- attributes.put(Queue.DURABLE, false);
+ attributes.put(ConfiguredObject.ID, UUID.randomUUID());
+ attributes.put(ConfiguredObject.NAME, nodeName);
+ attributes.put(ConfiguredObject.DURABLE, true);
if(lifetimePolicy instanceof DeleteOnNoLinks)
{
- attributes.put(Queue.LIFETIME_POLICY,
+ attributes.put(ConfiguredObject.LIFETIME_POLICY,
org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_NO_LINKS);
}
else if(lifetimePolicy instanceof DeleteOnNoLinksOrMessages)
{
- attributes.put(Queue.LIFETIME_POLICY,
+ attributes.put(ConfiguredObject.LIFETIME_POLICY,
org.apache.qpid.server.model.LifetimePolicy.IN_USE);
}
else if(lifetimePolicy instanceof DeleteOnClose)
{
- attributes.put(Queue.LIFETIME_POLICY,
- org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_CONNECTION_CLOSE);
+ attributes.put(ConfiguredObject.LIFETIME_POLICY,
+ org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_CREATING_LINK_CLOSE);
+ final CreatingLinkInfo linkInfo = new CreatingLinkInfoImpl(link.getRole() == Role.SENDER, link.getRemoteContainerId(), link.getName());
+ attributes.put("creatingLinkInfo", linkInfo);
}
else if(lifetimePolicy instanceof DeleteOnNoMessages)
{
- attributes.put(Queue.LIFETIME_POLICY,
+ attributes.put(ConfiguredObject.LIFETIME_POLICY,
org.apache.qpid.server.model.LifetimePolicy.IN_USE);
}
else
{
- attributes.put(Queue.LIFETIME_POLICY,
+ attributes.put(ConfiguredObject.LIFETIME_POLICY,
org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_CONNECTION_CLOSE);
}
return attributes;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java
index 86771ce..4057d48 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java
@@ -40,6 +40,7 @@ import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
import org.apache.qpid.server.protocol.v1_0.type.Binary;
import org.apache.qpid.server.protocol.v1_0.type.DeliveryState;
import org.apache.qpid.server.protocol.v1_0.type.Outcome;
+import org.apache.qpid.server.protocol.v1_0.type.Symbol;
import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
import org.apache.qpid.server.protocol.v1_0.type.messaging.AmqpSequenceSection;
import org.apache.qpid.server.protocol.v1_0.type.messaging.AmqpValueSection;
@@ -468,9 +469,20 @@ public class StandardReceivingLinkEndpoint extends AbstractReceivingLinkEndpoint
setDeliveryCount(attach.getInitialDeliveryCount());
- final ReceivingDestination destination = getSession().getReceivingDestination(attachTarget);
+ final ReceivingDestination destination = getSession().getReceivingDestination(getLink(), attachTarget);
target.setAddress(attachTarget.getAddress());
target.setDynamic(attachTarget.getDynamic());
+ if (Boolean.TRUE.equals(attachTarget.getDynamic()) && attachTarget.getDynamicNodeProperties() != null)
+ {
+ Map<Symbol, Object> dynamicNodeProperties = new HashMap<>();
+ if (attachTarget.getDynamicNodeProperties().containsKey(Session_1_0.LIFETIME_POLICY))
+ {
+ dynamicNodeProperties.put(Session_1_0.LIFETIME_POLICY,
+ attachTarget.getDynamicNodeProperties().get(Session_1_0.LIFETIME_POLICY));
+ }
+ source.setDynamicNodeProperties(dynamicNodeProperties);
+ }
+
target.setCapabilities(destination.getCapabilities());
target.setDurable(TerminusDurability.min(attachTarget.getDurable(),
getLink().getHighestSupportedTerminusDurability()));
@@ -520,7 +532,7 @@ public class StandardReceivingLinkEndpoint extends AbstractReceivingLinkEndpoint
Source source = (Source) attach.getSource();
Target target = getTarget();
- final ReceivingDestination destination = getSession().getReceivingDestination(getTarget());
+ final ReceivingDestination destination = getSession().getReceivingDestination(getLink(), getTarget());
target.setCapabilities(destination.getCapabilities());
setCapabilities(Arrays.asList(destination.getCapabilities()));
setDestination(destination);
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnClose.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnClose.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnClose.java
index e7676dd..f99ffe4 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnClose.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnClose.java
@@ -24,23 +24,13 @@
package org.apache.qpid.server.protocol.v1_0.type.messaging;
+import org.apache.qpid.server.protocol.v1_0.type.LifetimePolicy;
-import org.apache.qpid.server.protocol.v1_0.type.*;
-
-public class DeleteOnClose
- implements LifetimePolicy
- {
-
-
+public class DeleteOnClose implements LifetimePolicy
+{
@Override
public String toString()
{
- StringBuilder builder = new StringBuilder("DeleteOnClose{");
- final int origLength = builder.length();
-
- builder.append('}');
- return builder.toString();
+ return "DeleteOnClose{}";
}
-
-
- }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoLinks.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoLinks.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoLinks.java
index 7c1fa4d..8605d28 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoLinks.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoLinks.java
@@ -24,23 +24,13 @@
package org.apache.qpid.server.protocol.v1_0.type.messaging;
+import org.apache.qpid.server.protocol.v1_0.type.LifetimePolicy;
-import org.apache.qpid.server.protocol.v1_0.type.*;
-
-public class DeleteOnNoLinks
- implements LifetimePolicy
- {
-
-
+public class DeleteOnNoLinks implements LifetimePolicy
+{
@Override
public String toString()
{
- StringBuilder builder = new StringBuilder("DeleteOnNoLinks{");
- final int origLength = builder.length();
-
- builder.append('}');
- return builder.toString();
+ return "DeleteOnNoLinks{}";
}
-
-
- }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoLinksOrMessages.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoLinksOrMessages.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoLinksOrMessages.java
index 55e90aa..9e186b9 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoLinksOrMessages.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoLinksOrMessages.java
@@ -24,23 +24,13 @@
package org.apache.qpid.server.protocol.v1_0.type.messaging;
+import org.apache.qpid.server.protocol.v1_0.type.LifetimePolicy;
-import org.apache.qpid.server.protocol.v1_0.type.*;
-
-public class DeleteOnNoLinksOrMessages
- implements LifetimePolicy
- {
-
-
+public class DeleteOnNoLinksOrMessages implements LifetimePolicy
+{
@Override
public String toString()
{
- StringBuilder builder = new StringBuilder("DeleteOnNoLinksOrMessages{");
- final int origLength = builder.length();
-
- builder.append('}');
- return builder.toString();
+ return "DeleteOnNoLinksOrMessages{}";
}
-
-
- }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoMessages.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoMessages.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoMessages.java
index 0284439..27362bf 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoMessages.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/DeleteOnNoMessages.java
@@ -24,23 +24,13 @@
package org.apache.qpid.server.protocol.v1_0.type.messaging;
+import org.apache.qpid.server.protocol.v1_0.type.LifetimePolicy;
-import org.apache.qpid.server.protocol.v1_0.type.*;
-
-public class DeleteOnNoMessages
- implements LifetimePolicy
- {
-
-
+public class DeleteOnNoMessages implements LifetimePolicy
+{
@Override
public String toString()
{
- StringBuilder builder = new StringBuilder("DeleteOnNoMessages{");
- final int origLength = builder.length();
-
- builder.append('}');
- return builder.toString();
+ return "DeleteOnNoMessages{}";
}
-
-
- }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/Source.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/Source.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/Source.java
index 1b0c026..348f885 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/Source.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/Source.java
@@ -51,7 +51,7 @@ public class Source implements BaseSource
private Boolean _dynamic;
@CompositeTypeField
- private Map _dynamicNodeProperties;
+ private Map<Symbol, Object> _dynamicNodeProperties;
@CompositeTypeField
private DistributionMode _distributionMode;
@@ -118,12 +118,12 @@ public class Source implements BaseSource
_dynamic = dynamic;
}
- public Map getDynamicNodeProperties()
+ public Map<Symbol, Object> getDynamicNodeProperties()
{
return _dynamicNodeProperties;
}
- public void setDynamicNodeProperties(Map dynamicNodeProperties)
+ public void setDynamicNodeProperties(Map<Symbol, Object> dynamicNodeProperties)
{
_dynamicNodeProperties = dynamicNodeProperties;
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/Target.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/Target.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/Target.java
index 4e31aec..d0a564b 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/Target.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/type/messaging/Target.java
@@ -52,7 +52,7 @@ public class Target
private Boolean _dynamic;
@CompositeTypeField
- private Map _dynamicNodeProperties;
+ private Map<Symbol, Object> _dynamicNodeProperties;
@CompositeTypeField
private Symbol[] _capabilities;
@@ -107,12 +107,12 @@ public class Target
_dynamic = dynamic;
}
- public Map getDynamicNodeProperties()
+ public Map<Symbol, Object> getDynamicNodeProperties()
{
return _dynamicNodeProperties;
}
- public void setDynamicNodeProperties(Map dynamicNodeProperties)
+ public void setDynamicNodeProperties(Map<Symbol, Object> dynamicNodeProperties)
{
_dynamicNodeProperties = dynamicNodeProperties;
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreTestCase.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreTestCase.java b/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreTestCase.java
index d9755a6..2f0e1db 100644
--- a/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreTestCase.java
+++ b/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/store/LinkStoreTestCase.java
@@ -66,11 +66,12 @@ public abstract class LinkStoreTestCase extends QpidTestCase
_source.setExpiryPolicy(TerminusExpiryPolicy.CONNECTION_CLOSE);
_source.setFilter(Collections.singletonMap("foo", NoLocalFilter.INSTANCE));
_source.setOutcomes(new Accepted().getSymbol());
- _source.setDynamicNodeProperties(Collections.singletonMap("dynamicProperty", "dynamicPropertyValue"));
+ _source.setDynamicNodeProperties(Collections.singletonMap(Symbol.valueOf("dynamicProperty"),
+ "dynamicPropertyValue"));
_source.setTimeout(new UnsignedInteger(1));
_target.setTimeout(new UnsignedInteger(2));
- _target.setDynamicNodeProperties(Collections.singletonMap("targetDynamicProperty",
+ _target.setDynamicNodeProperties(Collections.singletonMap(Symbol.valueOf("targetDynamicProperty"),
"targetDynamicPropertyValue"));
_target.setDynamic(Boolean.TRUE);
_target.setExpiryPolicy(TerminusExpiryPolicy.LINK_DETACH);
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/BrokerAdmin.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/BrokerAdmin.java b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/BrokerAdmin.java
index 1a746fa..e4efc76 100644
--- a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/BrokerAdmin.java
+++ b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/BrokerAdmin.java
@@ -22,11 +22,14 @@ package org.apache.qpid.tests.protocol.v1_0;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
+import com.google.common.util.concurrent.ListenableFuture;
+
import org.apache.qpid.server.plugin.Pluggable;
public interface BrokerAdmin extends Pluggable
{
String TEST_QUEUE_NAME = "testQueue";
+ Long RESTART_TIMEOUT = Long.getLong("brokerAdmin.restart_timeout", 10000);
void beforeTestClass(final Class testClass);
void beforeTestMethod(final Class testClass, final Method method);
@@ -39,6 +42,9 @@ public interface BrokerAdmin extends Pluggable
void deleteQueue(String queueName);
void putMessageOnQueue(String queueName, String... messages);
+ boolean supportsRestart();
+ ListenableFuture<Void> restart();
+
enum PortType
{
ANONYMOUS_AMQP,
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/EmbeddedBrokerPerClassAdminImpl.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/EmbeddedBrokerPerClassAdminImpl.java b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/EmbeddedBrokerPerClassAdminImpl.java
index 54c5fb9..1240b94 100644
--- a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/EmbeddedBrokerPerClassAdminImpl.java
+++ b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/EmbeddedBrokerPerClassAdminImpl.java
@@ -35,6 +35,8 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import ch.qos.logback.classic.LoggerContext;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,6 +69,7 @@ public class EmbeddedBrokerPerClassAdminImpl implements BrokerAdmin
private Container<?> _broker;
private VirtualHostNode<?> _currentVirtualHostNode;
private String _currentWorkDirectory;
+ private boolean _isPersistentStore;
@Override
public void beforeTestClass(final Class testClass)
@@ -116,6 +119,7 @@ public class EmbeddedBrokerPerClassAdminImpl implements BrokerAdmin
final String virtualHostNodeName = testClass.getSimpleName() + "_" + method.getName();
final String storeType = System.getProperty("virtualhostnode.type");
+ _isPersistentStore = !"Memory".equals(storeType);
String storeDir = null;
if (System.getProperty("profile", "").startsWith("java-dby-mem"))
@@ -302,6 +306,30 @@ public class EmbeddedBrokerPerClassAdminImpl implements BrokerAdmin
}
@Override
+ public boolean supportsRestart()
+ {
+ return _isPersistentStore;
+ }
+
+ @Override
+ public ListenableFuture<Void> restart()
+ {
+ try
+ {
+ LOGGER.info("Stopping VirtualHostNode for restart");
+ _currentVirtualHostNode.stop();
+ LOGGER.info("Starting VirtualHostNode for restart");
+ _currentVirtualHostNode.start();
+ LOGGER.info("Restarting VirtualHostNode completed");
+ }
+ catch (Exception e)
+ {
+ return Futures.immediateFailedFuture(e);
+ }
+ return Futures.immediateFuture(null);
+ }
+
+ @Override
public String getType()
{
return "EMBEDDED_BROKER_PER_CLASS";
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/ExternalQpidBrokerAdminImpl.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/ExternalQpidBrokerAdminImpl.java b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/ExternalQpidBrokerAdminImpl.java
index 135feb9..7692f3e 100644
--- a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/ExternalQpidBrokerAdminImpl.java
+++ b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/ExternalQpidBrokerAdminImpl.java
@@ -22,6 +22,7 @@ package org.apache.qpid.tests.protocol.v1_0;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
+import com.google.common.util.concurrent.ListenableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -94,6 +95,18 @@ public class ExternalQpidBrokerAdminImpl implements BrokerAdmin
}
@Override
+ public boolean supportsRestart()
+ {
+ return false;
+ }
+
+ @Override
+ public ListenableFuture<Void> restart()
+ {
+ throw new UnsupportedOperationException("External Qpid Broker does not support restart.");
+ }
+
+ @Override
public String getType()
{
return "EXTERNAL_BROKER";
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/FrameTransport.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/FrameTransport.java b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/FrameTransport.java
index 419696e..14b28a6 100644
--- a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/FrameTransport.java
+++ b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/FrameTransport.java
@@ -208,7 +208,8 @@ public class FrameTransport implements AutoCloseable
PerformativeResponse response = (PerformativeResponse) getNextResponse();
if (!(response.getFrameBody() instanceof Close))
{
- throw new IllegalStateException("Unexpected response to connection Close");
+ throw new IllegalStateException(String.format(
+ "Unexpected response to connection Close. Expected Close got '%s'", response.getFrameBody()));
}
}
@@ -224,7 +225,8 @@ public class FrameTransport implements AutoCloseable
PerformativeResponse response = (PerformativeResponse) getNextResponse();
if (!(response.getFrameBody() instanceof Begin))
{
- throw new IllegalStateException("Unexpected response to connection Begin");
+ throw new IllegalStateException(String.format(
+ "Unexpected response to connection Begin. Expected Begin got '%s'", response.getFrameBody()));
}
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Utils.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Utils.java b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Utils.java
new file mode 100644
index 0000000..5ecc9e1
--- /dev/null
+++ b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Utils.java
@@ -0,0 +1,81 @@
+/*
+ * 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.tests.protocol.v1_0;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import java.net.InetSocketAddress;
+
+import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Attach;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Detach;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
+
+public class Utils
+{
+ public static boolean doesNodeExist(final InetSocketAddress brokerAddress,
+ final String nodeAddress) throws Exception
+ {
+ try (FrameTransport transport = new FrameTransport(brokerAddress))
+ {
+ transport.doBeginSession();
+
+ final boolean queueExists;
+ Attach validationAttach = new Attach();
+ validationAttach.setName("validationAttach");
+ validationAttach.setHandle(UnsignedInteger.ZERO);
+ validationAttach.setRole(Role.RECEIVER);
+ Source validationSource = new Source();
+ validationSource.setAddress(nodeAddress);
+ validationAttach.setSource(validationSource);
+ validationAttach.setTarget(new Target());
+ transport.sendPerformative(validationAttach);
+ PerformativeResponse validationResponse = (PerformativeResponse) transport.getNextResponse();
+ assertThat(validationResponse, is(notNullValue()));
+ assertThat(validationResponse.getFrameBody(), is(instanceOf(Attach.class)));
+ final Attach attachValidationResponse = (Attach) validationResponse.getFrameBody();
+ if (attachValidationResponse.getSource() != null)
+ {
+ queueExists = true;
+ Detach validationDetach = new Detach();
+ validationDetach.setHandle(validationAttach.getHandle());
+ validationDetach.setClosed(true);
+ transport.sendPerformative(validationDetach);
+ PerformativeResponse validationDetachResponse = (PerformativeResponse) transport.getNextResponse();
+ assertThat(validationDetachResponse, is(notNullValue()));
+ assertThat(validationDetachResponse.getFrameBody(), is(instanceOf(Detach.class)));
+ }
+ else
+ {
+ queueExists = false;
+ PerformativeResponse validationDetachResponse = (PerformativeResponse) transport.getNextResponse();
+ assertThat(validationDetachResponse, is(notNullValue()));
+ assertThat(validationDetachResponse.getFrameBody(), is(instanceOf(Detach.class)));
+ }
+ return queueExists;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/3bdc2e35/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/bindmapjms/TemporaryDestinationTest.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/bindmapjms/TemporaryDestinationTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/bindmapjms/TemporaryDestinationTest.java
new file mode 100644
index 0000000..74b2c11
--- /dev/null
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/bindmapjms/TemporaryDestinationTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.tests.protocol.v1_0.extensions.bindmapjms;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import java.net.InetSocketAddress;
+import java.util.Collections;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.qpid.server.protocol.v1_0.Session_1_0;
+import org.apache.qpid.server.protocol.v1_0.type.Symbol;
+import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.DeleteOnClose;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusExpiryPolicy;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Attach;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Flow;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
+import org.apache.qpid.tests.protocol.v1_0.BrokerAdmin;
+import org.apache.qpid.tests.protocol.v1_0.FrameTransport;
+import org.apache.qpid.tests.protocol.v1_0.PerformativeResponse;
+import org.apache.qpid.tests.protocol.v1_0.ProtocolTestBase;
+import org.apache.qpid.tests.protocol.v1_0.SpecificationTest;
+import org.apache.qpid.tests.protocol.v1_0.Utils;
+
+public class TemporaryDestinationTest extends ProtocolTestBase
+{
+ private InetSocketAddress _brokerAddress;
+
+ @Before
+ public void setUp()
+ {
+ _brokerAddress = getBrokerAdmin().getBrokerAddress(BrokerAdmin.PortType.ANONYMOUS_AMQP);
+ }
+
+ @Test
+ @SpecificationTest(section = "5.3",
+ description = "To create a node with the required lifecycle properties, establish a uniquely named sending link with "
+ + "the dynamic field of target set true, the expiry-policy field of target set to symbol “link-detach”, and the "
+ + "dynamic-node-properties field of target containing the “lifetime-policy” symbol key mapped to delete-on-close.")
+ public void deleteOnCloseWithLinkDetachForQueue() throws Exception
+ {
+ deleteOnCloseWithLinkDetach(new Symbol[]{Symbol.valueOf("temporary-queue")});
+ }
+
+ @Test
+ @SpecificationTest(section = "5.3",
+ description = "To create a node with the required lifecycle properties, establish a uniquely named sending link with "
+ + "the dynamic field of target set true, the expiry-policy field of target set to symbol “link-detach”, and the "
+ + "dynamic-node-properties field of target containing the “lifetime-policy” symbol key mapped to delete-on-close.")
+ public void deleteOnCloseWithLinkDetachForTopic() throws Exception
+ {
+ deleteOnCloseWithLinkDetach(new Symbol[]{Symbol.valueOf("temporary-topic")});
+ }
+
+ private void deleteOnCloseWithLinkDetach(final Symbol[] targetCapabilities) throws Exception
+ {
+ String newTemporaryNodeAddress = null;
+
+ try (FrameTransport transport = new FrameTransport(_brokerAddress))
+ {
+ transport.doBeginSession();
+
+ Attach attach = new Attach();
+ attach.setName("testSendingLink");
+ attach.setHandle(UnsignedInteger.ZERO);
+ attach.setRole(Role.SENDER);
+ attach.setInitialDeliveryCount(UnsignedInteger.ZERO);
+
+ attach.setSource(new Source());
+
+ Target target = new Target();
+ target.setDynamicNodeProperties(Collections.singletonMap(Session_1_0.LIFETIME_POLICY, new DeleteOnClose()));
+ target.setExpiryPolicy(TerminusExpiryPolicy.LINK_DETACH);
+ target.setDynamic(true);
+ target.setCapabilities(targetCapabilities);
+ attach.setTarget(target);
+
+ transport.sendPerformative(attach);
+
+ PerformativeResponse response = (PerformativeResponse) transport.getNextResponse();
+ assertThat(response, is(notNullValue()));
+ assertThat(response.getFrameBody(), is(instanceOf(Attach.class)));
+ final Attach attachResponse = (Attach) response.getFrameBody();
+ assertThat(attachResponse.getSource(), is(notNullValue()));
+ assertThat(attachResponse.getTarget(), is(notNullValue()));
+
+ newTemporaryNodeAddress = ((Target) attachResponse.getTarget()).getAddress();
+ assertThat(newTemporaryNodeAddress, is(notNullValue()));
+
+ assertThat(Utils.doesNodeExist(_brokerAddress, newTemporaryNodeAddress), is(true));
+
+ final PerformativeResponse flowResponse = ((PerformativeResponse) transport.getNextResponse());
+ if (flowResponse != null)
+ {
+ assertThat(flowResponse.getFrameBody(), is(instanceOf(Flow.class)));
+ }
+
+ transport.doCloseConnection();
+ }
+
+ try (FrameTransport transport = new FrameTransport(_brokerAddress))
+ {
+ transport.doBeginSession();
+ assertThat(Utils.doesNodeExist(_brokerAddress, newTemporaryNodeAddress), is(false));
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org