QPID-6834: allow delete-on-close in queue policies
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1713025 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/qpid/cpp/src/qpid/broker/QueueSettings.cpp b/qpid/cpp/src/qpid/broker/QueueSettings.cpp
index 8de8539..29d482d 100644
--- a/qpid/cpp/src/qpid/broker/QueueSettings.cpp
+++ b/qpid/cpp/src/qpid/broker/QueueSettings.cpp
@@ -60,6 +60,7 @@
const std::string PAGE_FACTOR("qpid.page_factor");
const std::string FILTER("qpid.filter");
const std::string LIFETIME_POLICY("qpid.lifetime-policy");
+const std::string DELETE_ON_CLOSE_KEY("delete-on-close");
const std::string DELETE_IF_UNUSED_KEY("delete-if-unused");
const std::string DELETE_IF_UNUSED_AND_EMPTY_KEY("delete-if-unused-and-empty");
const std::string MANUAL("manual");
@@ -234,6 +235,9 @@
autodelete = true;
} else if (value.asString() == MANUAL) {
autodelete = false;
+ } else if (value.asString() == DELETE_ON_CLOSE_KEY) {
+ lifetime = DELETE_ON_CLOSE;
+ autodelete = true;
} else {
QPID_LOG(warning, "Invalid value for " << LIFETIME_POLICY << ": " << value);
}
diff --git a/qpid/cpp/src/qpid/broker/amqp/Session.cpp b/qpid/cpp/src/qpid/broker/amqp/Session.cpp
index 2f793ae..e0ca8f5 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Session.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/Session.cpp
@@ -231,6 +231,11 @@
private:
};
+bool Session::ResolvedNode::trackControllingLink() const
+{
+ return created && (properties.trackControllingLink() || (queue && queue->getSettings().lifetime == QueueSettings::DELETE_ON_CLOSE));
+}
+
Session::Session(pn_session_t* s, Connection& c, qpid::sys::OutputControl& o)
: ManagedSession(c.getBroker(), c, (boost::format("%1%") % s).str()), session(s), connection(c), out(o), deleted(false),
authorise(connection.getUserId(), connection.getBroker().getAcl()),
@@ -306,6 +311,7 @@
std::pair<boost::shared_ptr<Queue>, boost::shared_ptr<Topic> > result = nodePolicy->create(name, connection);
node.queue = result.first;
node.topic = result.second;
+ node.created = node.queue || node.topic;
if (node.topic) node.exchange = node.topic->getExchange();
if (node.queue) {
@@ -478,10 +484,10 @@
source = sourceAddress;
}
if (node.queue) {
- boost::shared_ptr<Incoming> q(new IncomingToQueue(connection.getBroker(), *this, node.queue, link, source, node.created && node.properties.trackControllingLink()));
+ boost::shared_ptr<Incoming> q(new IncomingToQueue(connection.getBroker(), *this, node.queue, link, source, node.trackControllingLink()));
incoming[link] = q;
} else if (node.exchange) {
- boost::shared_ptr<Incoming> e(new IncomingToExchange(connection.getBroker(), *this, node.exchange, link, source, node.created && node.properties.trackControllingLink()));
+ boost::shared_ptr<Incoming> e(new IncomingToExchange(connection.getBroker(), *this, node.exchange, link, source, node.trackControllingLink()));
incoming[link] = e;
} else if (node.relay) {
boost::shared_ptr<Incoming> in(new IncomingToRelay(link, connection.getBroker(), *this, source, name, pn_link_name(link), node.relay));
@@ -523,7 +529,7 @@
if (type == CONSUMER && node.queue->hasExclusiveOwner() && !node.queue->isExclusiveOwner(this)) {
throw Exception(qpid::amqp::error_conditions::PRECONDITION_FAILED, std::string("Cannot consume from exclusive queue ") + node.queue->getName());
}
- boost::shared_ptr<Outgoing> q(new OutgoingFromQueue(connection.getBroker(), name, target, node.queue, link, *this, out, type, false, node.created && node.properties.trackControllingLink()));
+ boost::shared_ptr<Outgoing> q(new OutgoingFromQueue(connection.getBroker(), name, target, node.queue, link, *this, out, type, false, node.trackControllingLink()));
q->init();
filter.apply(q);
outgoing[link] = q;
diff --git a/qpid/cpp/src/qpid/broker/amqp/Session.h b/qpid/cpp/src/qpid/broker/amqp/Session.h
index 8337606..e48d563 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Session.h
+++ b/qpid/cpp/src/qpid/broker/amqp/Session.h
@@ -134,6 +134,7 @@
NodeProperties properties;
bool created;
ResolvedNode(bool isDynamic) : properties(isDynamic), created(false) {}
+ bool trackControllingLink() const;
};
ResolvedNode resolve(const std::string name, pn_terminus_t* terminus, bool incoming);
