You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2013/07/10 15:47:06 UTC
svn commit: r1501768 - in /qpid/trunk/qpid/cpp/src: qpid/amqp/ qpid/broker/
qpid/broker/amqp/ qpid/ha/ qpid/messaging/amqp/ tests/
Author: gsim
Date: Wed Jul 10 13:47:05 2013
New Revision: 1501768
URL: http://svn.apache.org/r1501768
Log:
QPID-4976: support standard lifetime policies
Modified:
qpid/trunk/qpid/cpp/src/qpid/amqp/descriptors.h
qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h
qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.h
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/DataReader.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.h
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.h
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h
qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp
qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/amqp/descriptors.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/amqp/descriptors.h?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/amqp/descriptors.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/amqp/descriptors.h Wed Jul 10 13:47:05 2013
@@ -89,6 +89,18 @@ const uint64_t SELECTOR_FILTER_CODE(0x00
const uint64_t XQUERY_FILTER_CODE(0x0000468C00000005ULL);
}
+namespace lifetime_policy {
+const std::string DELETE_ON_CLOSE_SYMBOL("amqp:delete-on-close:list");
+const std::string DELETE_ON_NO_LINKS_SYMBOL("amqp:delete-on-no-links:list");
+const std::string DELETE_ON_NO_MESSAGES_SYMBOL("amqp:delete-on-no-messages:list");
+const std::string DELETE_ON_NO_LINKS_OR_MESSAGES_SYMBOL("amqp:delete-on-no-links-or-messages:list");
+
+const uint64_t DELETE_ON_CLOSE_CODE(0x2B);
+const uint64_t DELETE_ON_NO_LINKS_CODE(0x2C);
+const uint64_t DELETE_ON_NO_MESSAGES_CODE(0x2D);
+const uint64_t DELETE_ON_NO_LINKS_OR_MESSAGES_CODE(0x2E);
+}
+
namespace error_conditions {
//note these are not actually descriptors
const std::string INTERNAL_ERROR("amqp:internal-error");
@@ -97,6 +109,8 @@ const std::string UNAUTHORIZED_ACCESS("a
const std::string DECODE_ERROR("amqp:decode-error");
const std::string NOT_ALLOWED("amqp:not-allowed");
const std::string RESOURCE_LIMIT_EXCEEDED("amqp:resource-limit-exceeded");
+const std::string RESOURCE_DELETED("amqp:resource-deleted");
+const std::string PRECONDITION_FAILED("amqp:precondition-failed");
}
}} // namespace qpid::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.cpp Wed Jul 10 13:47:05 2013
@@ -48,11 +48,11 @@ bool LossyQueue::checkDepth(const QueueD
<< name << ": size=" << increment.getSize() << ", max-size=" << settings.maxDepth.getSize()));
}
- while (settings.maxDepth && (current + increment > settings.maxDepth)) {
+ while (settings.maxDepth && (settings.maxDepth - current < increment)) {
QPID_LOG(debug, "purging " << name << ": current depth is [" << current << "], max depth is [" << settings.maxDepth << "], new message has size " << increment.getSize());
qpid::sys::Mutex::ScopedUnlock u(messageLock);
//TODO: arguably we should try and purge expired messages first but that is potentially expensive
- if (remove(1, settings.priorities ? boost::bind(&isLowerPriorityThan, message.getPriority(), _1) : MessagePredicate(), MessageFunctor(), PURGE)) {
+ if (remove(1, settings.priorities ? boost::bind(&isLowerPriorityThan, message.getPriority(), _1) : MessagePredicate(), MessageFunctor(), PURGE, false)) {
if (mgmtObject) {
mgmtObject->inc_discardsRing(1);
if (brokerMgmtObject)
@@ -65,7 +65,7 @@ bool LossyQueue::checkDepth(const QueueD
break;
}
}
- if (settings.maxDepth && (current + increment > settings.maxDepth)) {
+ if (settings.maxDepth && (settings.maxDepth - current < increment)) {
//will only be the case where we were unable to purge another
//message, which should only be the case if we are purging
//based on priority and there was no message with a lower (or
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.cpp Wed Jul 10 13:47:05 2013
@@ -51,7 +51,7 @@ void Lvq::push(Message& message, bool is
brokerMgmtObject->inc_discardsLvq();
}
}
- observeDequeue(old, locker);
+ observeDequeue(old, locker, 0/*can't be empty, so no need to check autodelete*/);
}
}
copy.notify();
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Wed Jul 10 13:47:05 2013
@@ -175,8 +175,6 @@ Queue::Queue(const string& _name, const
name(_name),
store(_store),
owner(0),
- consumerCount(0),
- browserCount(0),
exclusive(0),
messages(new MessageDeque()),
persistenceId(0),
@@ -188,8 +186,8 @@ Queue::Queue(const string& _name, const
allocator(new FifoDistributor( *messages )),
redirectSource(false)
{
- if (settings.maxDepth.hasCount()) current.setCount(0);
- if (settings.maxDepth.hasSize()) current.setSize(0);
+ current.setCount(0);//always track depth in messages
+ if (settings.maxDepth.hasSize()) current.setSize(0);//track depth in bytes only if policy requires it
if (settings.traceExcludes.size()) {
split(traceExclude, settings.traceExcludes, ", ");
}
@@ -340,6 +338,7 @@ void Queue::release(const QueueCursor& p
bool Queue::dequeueMessageAt(const SequenceNumber& position)
{
+ ScopedAutoDelete autodelete(*this);
boost::intrusive_ptr<PersistableMessage> pmsg;
{
Mutex::ScopedLock locker(messageLock);
@@ -348,7 +347,7 @@ bool Queue::dequeueMessageAt(const Seque
Message* msg = messages->find(position, &cursor);
if (msg) {
if (msg->isPersistent()) pmsg = msg->getPersistentContext();
- observeDequeue(*msg, locker);
+ observeDequeue(*msg, locker, settings.autodelete ? &autodelete : 0);
messages->deleted(cursor);
} else {
QPID_LOG(debug, "Could not dequeue message at " << position << "; no such message");
@@ -385,6 +384,7 @@ bool Queue::getNextMessage(Message& m, C
{
if (!checkNotDeleted(c)) return false;
QueueListeners::NotificationSet set;
+ ScopedAutoDelete autodelete(*this);
while (true) {
//TODO: reduce lock scope
Mutex::ScopedLock locker(messageLock);
@@ -393,7 +393,7 @@ bool Queue::getNextMessage(Message& m, C
if (msg) {
if (msg->hasExpired()) {
QPID_LOG(debug, "Message expired from queue '" << name << "'");
- observeDequeue(*msg, locker);
+ observeDequeue(*msg, locker, settings.autodelete ? &autodelete : 0);
//ERROR: don't hold lock across call to store!!
if (msg->isPersistent()) dequeueFromStore(msg->getPersistentContext());
if (mgmtObject) {
@@ -490,14 +490,31 @@ bool Queue::find(SequenceNumber pos, Mes
return false;
}
+void Queue::markInUse(bool controlling)
+{
+ Mutex::ScopedLock locker(messageLock);
+ if (controlling) users.addLifecycleController();
+ else users.addOther();
+}
+
+void Queue::releaseFromUse(bool controlling)
+{
+ if (controlling) {
+ {
+ Mutex::ScopedLock locker(messageLock);
+ users.removeLifecycleController();
+ }
+ scheduleAutoDelete();
+ } else {
+ Mutex::ScopedLock locker(messageLock);
+ users.removeOther();
+ }
+}
+
void Queue::consume(Consumer::shared_ptr c, bool requestExclusive)
{
{
Mutex::ScopedLock locker(messageLock);
- // NOTE: consumerCount is actually a count of all
- // subscriptions, both acquiring and non-acquiring (browsers).
- // Check for exclusivity of acquiring consumers.
- size_t acquiringConsumers = consumerCount - browserCount;
if (c->preAcquires()) {
if(settings.isBrowseOnly) {
throw NotAllowedException(
@@ -509,7 +526,7 @@ void Queue::consume(Consumer::shared_ptr
QPID_MSG("Queue " << getName()
<< " has an exclusive consumer. No more consumers allowed."));
} else if(requestExclusive) {
- if(acquiringConsumers) {
+ if(users.hasConsumers()) {
throw ResourceLockedException(
QPID_MSG("Queue " << getName()
<< " already has consumers. Exclusive access denied."));
@@ -517,13 +534,11 @@ void Queue::consume(Consumer::shared_ptr
exclusive = c->getSession();
}
}
- }
- else if(c->isCounted()) {
- browserCount++;
+ users.addConsumer();
+ } else if(c->isCounted()) {
+ users.addBrowser();
}
if(c->isCounted()) {
- consumerCount++;
-
//reset auto deletion timer if necessary
if (settings.autoDeleteDelay && autoDeleteTask) {
autoDeleteTask->cancel();
@@ -542,14 +557,24 @@ void Queue::cancel(Consumer::shared_ptr
removeListener(c);
if(c->isCounted())
{
- Mutex::ScopedLock locker(messageLock);
- consumerCount--;
- if (!c->preAcquires()) browserCount--;
- if(exclusive) exclusive = 0;
- observeConsumerRemove(*c, locker);
- }
- if (mgmtObject != 0 && c->isCounted()) {
- mgmtObject->dec_consumerCount();
+ bool unused;
+ {
+ Mutex::ScopedLock locker(messageLock);
+ if (c->preAcquires()) {
+ users.removeConsumer();
+ if (exclusive) exclusive = 0;
+ } else {
+ users.removeBrowser();
+ }
+ observeConsumerRemove(*c, locker);
+ unused = !users.isUsed();
+ }
+ if (mgmtObject != 0) {
+ mgmtObject->dec_consumerCount();
+ }
+ if (unused && settings.autodelete) {
+ scheduleAutoDelete();
+ }
}
}
@@ -564,7 +589,7 @@ void Queue::purgeExpired(sys::Duration l
dequeueSincePurge -= count;
int seconds = int64_t(lapse)/qpid::sys::TIME_SEC;
if (seconds == 0 || count / seconds < 1) {
- uint32_t count = remove(0, boost::bind(&Message::hasExpired, _1), 0, CONSUMER);
+ uint32_t count = remove(0, boost::bind(&Message::hasExpired, _1), 0, CONSUMER, settings.autodelete);
QPID_LOG(debug, "Purged " << count << " expired messages from " << getName());
//
// Report the count of discarded-by-ttl messages
@@ -671,8 +696,9 @@ namespace {
}
} // end namespace
-uint32_t Queue::remove(const uint32_t maxCount, MessagePredicate p, MessageFunctor f, SubscriptionType type)
+uint32_t Queue::remove(const uint32_t maxCount, MessagePredicate p, MessageFunctor f, SubscriptionType type, bool triggerAutoDelete)
{
+ ScopedAutoDelete autodelete(*this);
std::deque<Message> removed;
{
QueueCursor c(type);
@@ -686,7 +712,7 @@ uint32_t Queue::remove(const uint32_t ma
//don't actually acquire, just act as if we did
observeAcquire(*m, locker);
}
- observeDequeue(*m, locker);
+ observeDequeue(*m, locker, triggerAutoDelete ? &autodelete : 0);
removed.push_back(*m);//takes a copy of the message
if (!messages->deleted(c)) {
QPID_LOG(warning, "Failed to correctly remove message from " << name << "; state is not consistent!");
@@ -726,7 +752,7 @@ uint32_t Queue::purge(const uint32_t qty
const qpid::types::Variant::Map *filter)
{
std::auto_ptr<MessageFilter> mf(MessageFilter::create(filter));
- uint32_t count = remove(qty, boost::bind(&MessageFilter::match, mf.get(), _1), boost::bind(&reroute, dest, _1), CONSUMER/*?*/);
+ uint32_t count = remove(qty, boost::bind(&MessageFilter::match, mf.get(), _1), boost::bind(&reroute, dest, _1), CONSUMER/*?*/, settings.autodelete);
if (mgmtObject && count) {
mgmtObject->inc_acquires(count);
@@ -752,7 +778,7 @@ uint32_t Queue::move(const Queue::shared
const qpid::types::Variant::Map *filter)
{
std::auto_ptr<MessageFilter> mf(MessageFilter::create(filter));
- return remove(qty, boost::bind(&MessageFilter::match, mf.get(), _1), boost::bind(&moveTo, destq, _1), CONSUMER/*?*/);
+ return remove(qty, boost::bind(&MessageFilter::match, mf.get(), _1), boost::bind(&moveTo, destq, _1), CONSUMER/*?*/, settings.autodelete);
}
void Queue::push(Message& message, bool /*isRecovery*/)
@@ -779,15 +805,41 @@ uint32_t Queue::getMessageCount() const
uint32_t Queue::getConsumerCount() const
{
Mutex::ScopedLock locker(messageLock);
- return consumerCount;
+ return users.getSubscriberCount();
}
bool Queue::canAutoDelete() const
{
Mutex::ScopedLock locker(messageLock);
- return settings.autodelete && !consumerCount && !owner;
+ return !deleted && checkAutoDelete(locker);
+}
+
+bool Queue::checkAutoDelete(const Mutex::ScopedLock& lock) const
+{
+ if (settings.autodelete) {
+ switch (settings.lifetime) {
+ case QueueSettings::DELETE_IF_UNUSED:
+ return isUnused(lock);
+ case QueueSettings::DELETE_IF_EMPTY:
+ return !users.isInUseByController() && isEmpty(lock);
+ case QueueSettings::DELETE_IF_UNUSED_AND_EMPTY:
+ return isUnused(lock) && isEmpty(lock);
+ case QueueSettings::DELETE_ON_CLOSE:
+ return !users.isInUseByController();
+ }
+ }
+ return false;
+}
+
+bool Queue::isUnused(const Mutex::ScopedLock&) const
+{
+ return !owner && !users.isUsed();;
}
+bool Queue::isEmpty(const Mutex::ScopedLock&) const
+{
+ return current.getCount() == 0;
+}
/*
* return true if enqueue succeeded and message should be made
* available; returning false will result in the message being dropped
@@ -853,8 +905,9 @@ void Queue::dequeueCommited(const Messag
//store and will not be available for delivery. The only action
//required is to ensure the observers are notified and the
//management stats are correctly decremented
+ ScopedAutoDelete autodelete(*this);
Mutex::ScopedLock locker(messageLock);
- observeDequeue(msg, locker);
+ observeDequeue(msg, locker, settings.autodelete ? &autodelete : 0);
if (mgmtObject != 0) {
mgmtObject->inc_msgTxnDequeues();
mgmtObject->inc_byteTxnDequeues(msg.getContentSize());
@@ -874,6 +927,7 @@ void Queue::dequeue(TransactionContext*
{
ScopedUse u(barrier);
if (!u.acquired) return;
+ ScopedAutoDelete autodelete(*this);
boost::intrusive_ptr<PersistableMessage> pmsg;
{
Mutex::ScopedLock locker(messageLock);
@@ -881,7 +935,7 @@ void Queue::dequeue(TransactionContext*
if (msg) {
if (msg->isPersistent()) pmsg = msg->getPersistentContext();
if (!ctxt) {
- observeDequeue(*msg, locker);
+ observeDequeue(*msg, locker, settings.autodelete ? &autodelete : 0);
messages->deleted(cursor);//message pointer not valid after this
}
} else {
@@ -895,11 +949,12 @@ void Queue::dequeue(TransactionContext*
void Queue::dequeueCommitted(const QueueCursor& cursor)
{
+ ScopedAutoDelete autodelete(*this);
Mutex::ScopedLock locker(messageLock);
Message* msg = messages->find(cursor);
if (msg) {
const uint64_t contentSize = msg->getContentSize();
- observeDequeue(*msg, locker);
+ observeDequeue(*msg, locker, settings.autodelete ? &autodelete : 0);
if (mgmtObject != 0) {
mgmtObject->inc_msgTxnDequeues();
mgmtObject->inc_byteTxnDequeues(contentSize);
@@ -920,7 +975,7 @@ void Queue::dequeueCommitted(const Queue
* Updates policy and management when a message has been dequeued,
* Requires messageLock be held by caller.
*/
-void Queue::observeDequeue(const Message& msg, const Mutex::ScopedLock&)
+void Queue::observeDequeue(const Message& msg, const Mutex::ScopedLock& lock, ScopedAutoDelete* autodelete)
{
current -= QueueDepth(1, msg.getContentSize());
mgntDeqStats(msg, mgmtObject, brokerMgmtObject);
@@ -931,6 +986,7 @@ void Queue::observeDequeue(const Message
QPID_LOG(warning, "Exception on notification of dequeue for queue " << getName() << ": " << e.what());
}
}
+ if (autodelete && isEmpty(lock)) autodelete->check(lock);
}
/** updates queue observers when a message has become unavailable for transfer.
@@ -1053,9 +1109,10 @@ void Queue::abandoned(const Message& mes
void Queue::destroyed()
{
unbind(broker->getExchanges());
- remove(0, 0, boost::bind(&Queue::abandoned, this, _1), REPLICATOR/*even acquired message are treated as abandoned*/);
+ remove(0, 0, boost::bind(&Queue::abandoned, this, _1), REPLICATOR/*even acquired message are treated as abandoned*/, false);
if (alternateExchange.get()) {
alternateExchange->decAlternateUsers();
+ alternateExchange.reset();
}
if (store) {
@@ -1170,33 +1227,12 @@ boost::shared_ptr<Exchange> Queue::getAl
return alternateExchange;
}
-void tryAutoDeleteImpl(Broker& broker, Queue::shared_ptr queue, const std::string& connectionId, const std::string& userId)
-{
- if (broker.getQueues().destroyIf(queue->getName(),
- boost::bind(boost::mem_fn(&Queue::canAutoDelete), queue))) {
- if (broker.getAcl())
- broker.getAcl()->recordDestroyQueue(queue->getName());
-
- QPID_LOG_CAT(debug, model, "Auto-delete queue: " << queue->getName()
- << " user:" << userId
- << " rhost:" << connectionId );
- queue->destroyed();
- } else {
- QPID_LOG_CAT(debug, model, "Auto-delete queue could not be deleted: " << queue->getName()
- << " user:" << userId
- << " rhost:" << connectionId );
- }
-}
-
struct AutoDeleteTask : qpid::sys::TimerTask
{
- Broker& broker;
Queue::shared_ptr queue;
- std::string connectionId;
- std::string userId;
- AutoDeleteTask(Broker& b, Queue::shared_ptr q, const std::string& cId, const std::string& uId, AbsTime fireTime)
- : qpid::sys::TimerTask(fireTime, "DelayedAutoDeletion:"+q->getName()), broker(b), queue(q), connectionId(cId), userId(uId) {}
+ AutoDeleteTask(Queue::shared_ptr q, AbsTime fireTime)
+ : qpid::sys::TimerTask(fireTime, "DelayedAutoDeletion:"+q->getName()), queue(q) {}
void fire()
{
@@ -1204,34 +1240,56 @@ struct AutoDeleteTask : qpid::sys::Timer
//created, but then became unused again before the task fired;
//in this case ignore this request as there will have already
//been a later task added
- tryAutoDeleteImpl(broker, queue, connectionId, userId);
+ queue->tryAutoDelete();
}
};
-void Queue::tryAutoDelete(Broker& broker, Queue::shared_ptr queue, const std::string& connectionId, const std::string& userId)
+void Queue::scheduleAutoDelete()
+{
+ if (canAutoDelete()) {
+ if (settings.autoDeleteDelay) {
+ AbsTime time(now(), Duration(settings.autoDeleteDelay * TIME_SEC));
+ autoDeleteTask = boost::intrusive_ptr<qpid::sys::TimerTask>(new AutoDeleteTask(shared_from_this(), time));
+ broker->getTimer().add(autoDeleteTask);
+ QPID_LOG(debug, "Timed auto-delete for " << getName() << " initiated");
+ } else {
+ tryAutoDelete();
+ }
+ }
+}
+
+void Queue::tryAutoDelete()
{
- if (queue->settings.autoDeleteDelay && queue->canAutoDelete()) {
- AbsTime time(now(), Duration(queue->settings.autoDeleteDelay * TIME_SEC));
- queue->autoDeleteTask = boost::intrusive_ptr<qpid::sys::TimerTask>(new AutoDeleteTask(broker, queue, connectionId, userId, time));
- broker.getTimer().add(queue->autoDeleteTask);
- QPID_LOG(debug, "Timed auto-delete for " << queue->getName() << " initiated");
+ if (broker->getQueues().destroyIf(name, boost::bind(boost::mem_fn(&Queue::canAutoDelete), shared_from_this()))) {
+ if (broker->getAcl())
+ broker->getAcl()->recordDestroyQueue(name);
+
+ QPID_LOG_CAT(debug, model, "Auto-delete queue deleted: " << name << " (" << deleted << ")");
+ destroyed();
} else {
- tryAutoDeleteImpl(broker, queue, connectionId, userId);
+ QPID_LOG_CAT(debug, model, "Auto-delete queue could not be deleted: " << name);
}
}
bool Queue::isExclusiveOwner(const OwnershipToken* const o) const
{
- Mutex::ScopedLock locker(ownershipLock);
+ Mutex::ScopedLock locker(messageLock);
return o == owner;
}
void Queue::releaseExclusiveOwnership()
{
- Mutex::ScopedLock locker(ownershipLock);
- owner = 0;
- if (mgmtObject) {
- mgmtObject->set_exclusive(false);
+ bool unused;
+ {
+ Mutex::ScopedLock locker(messageLock);
+ owner = 0;
+ if (mgmtObject) {
+ mgmtObject->set_exclusive(false);
+ }
+ unused = !users.isUsed();
+ }
+ if (unused && settings.autodelete) {
+ scheduleAutoDelete();
}
}
@@ -1241,7 +1299,7 @@ bool Queue::setExclusiveOwner(const Owne
if (settings.autoDeleteDelay && autoDeleteTask) {
autoDeleteTask->cancel();
}
- Mutex::ScopedLock locker(ownershipLock);
+ Mutex::ScopedLock locker(messageLock);
if (owner) {
return false;
} else {
@@ -1255,7 +1313,7 @@ bool Queue::setExclusiveOwner(const Owne
bool Queue::hasExclusiveOwner() const
{
- Mutex::ScopedLock locker(ownershipLock);
+ Mutex::ScopedLock locker(messageLock);
return owner != 0;
}
@@ -1388,7 +1446,7 @@ struct After {
void Queue::setPosition(SequenceNumber n) {
Mutex::ScopedLock locker(messageLock);
if (n < sequence) {
- remove(0, After(n), MessagePredicate(), BROWSER);
+ remove(0, After(n), MessagePredicate(), BROWSER, false);
}
sequence = n;
QPID_LOG(debug, "Set position to " << sequence << " on " << getName());
@@ -1450,6 +1508,12 @@ bool Queue::checkNotDeleted(const Consum
return !deleted;
}
+bool Queue::isDeleted() const
+{
+ Mutex::ScopedLock lock(messageLock);
+ return deleted;
+}
+
void Queue::addObserver(boost::shared_ptr<QueueObserver> observer)
{
Mutex::ScopedLock lock(messageLock);
@@ -1495,6 +1559,7 @@ void Queue::setDequeueSincePurge(uint32_
void Queue::reject(const QueueCursor& cursor)
{
+ ScopedAutoDelete autodelete(*this);
Exchange::shared_ptr alternate = getAlternateExchange();
Message copy;
boost::intrusive_ptr<PersistableMessage> pmsg;
@@ -1505,7 +1570,7 @@ void Queue::reject(const QueueCursor& cu
if (alternate) copy = *message;
if (message->isPersistent()) pmsg = message->getPersistentContext();
countRejected();
- observeDequeue(*message, locker);
+ observeDequeue(*message, locker, settings.autodelete ? &autodelete : 0);
messages->deleted(cursor);
} else {
return;
@@ -1526,7 +1591,7 @@ void Queue::reject(const QueueCursor& cu
bool Queue::checkDepth(const QueueDepth& increment, const Message&)
{
- if (current && (settings.maxDepth - current < increment)) {
+ if (settings.maxDepth && (settings.maxDepth - current < increment)) {
if (mgmtObject) {
mgmtObject->inc_discardsOverflow();
if (brokerMgmtObject)
@@ -1619,6 +1684,29 @@ void Queue::setMgmtRedirectState( std::s
mgmtObject->set_redirectSource(isSrc);
}
}
+Queue::QueueUsers::QueueUsers() : consumers(0), browsers(0), others(0), controller(false) {}
+void Queue::QueueUsers::addConsumer() { ++consumers; }
+void Queue::QueueUsers::addBrowser() { ++browsers; }
+void Queue::QueueUsers::addLifecycleController() { assert(!controller); controller = true; }
+void Queue::QueueUsers::addOther(){ ++others; }
+void Queue::QueueUsers::removeConsumer() { assert(consumers > 0); --consumers; }
+void Queue::QueueUsers::removeBrowser() { assert(browsers > 0); --browsers; }
+void Queue::QueueUsers::removeLifecycleController() { assert(controller); controller = false; }
+void Queue::QueueUsers::removeOther() { assert(others > 0); --others; }
+bool Queue::QueueUsers::isInUseByController() const { return controller; }
+bool Queue::QueueUsers::isUsed() const { return controller || consumers || browsers || others; }
+uint32_t Queue::QueueUsers::getSubscriberCount() const { return consumers + browsers; }
+bool Queue::QueueUsers::hasConsumers() const { return consumers; }
+
+Queue::ScopedAutoDelete::ScopedAutoDelete(Queue& q) : queue(q), eligible(false) {}
+void Queue::ScopedAutoDelete::check(const sys::Mutex::ScopedLock& lock)
+{
+ eligible = queue.checkAutoDelete(lock);
+}
+Queue::ScopedAutoDelete::~ScopedAutoDelete()
+{
+ if (eligible) queue.scheduleAutoDelete();
+}
}}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Wed Jul 10 13:47:05 2013
@@ -87,6 +87,7 @@ class Queue : public boost::enable_share
typedef boost::shared_ptr<Queue> shared_ptr;
protected:
+ friend struct AutoDeleteTask;
struct UsageBarrier
{
Queue& parent;
@@ -119,6 +120,54 @@ class Queue : public boost::enable_share
void rollback() throw();
};
+ /**
+ * This class tracks whether a queue is in use and how it is being
+ * used.
+ */
+ class QueueUsers
+ {
+ public:
+ QueueUsers();
+ void addConsumer();
+ void addBrowser();
+ void addOther();
+ void removeConsumer();
+ void removeBrowser();
+ void addLifecycleController();
+ void removeLifecycleController();
+ void removeOther();
+ bool isUsed() const;
+ uint32_t getSubscriberCount() const;
+ bool hasConsumers() const;
+ bool isInUseByController() const;
+ private:
+ uint32_t consumers;
+ uint32_t browsers;
+ uint32_t others;
+ bool controller;
+ };
+
+ /**
+ * This class is used to check - and if necessary trigger -
+ * autodeletion when removing messages, as this could cause the
+ * queue to become empty (which is one possible trigger for
+ * autodeletion).
+ *
+ * The constructor and descructor should be called outside the
+ * message lock. The check method should be called while holding
+ * the message lock.
+ */
+ class ScopedAutoDelete
+ {
+ public:
+ ScopedAutoDelete(Queue& q);
+ void check(const sys::Mutex::ScopedLock& lock);
+ ~ScopedAutoDelete();
+ private:
+ Queue& queue;
+ bool eligible;
+ };
+
typedef std::set< boost::shared_ptr<QueueObserver> > Observers;
enum ConsumeCode {NO_MESSAGES=0, CANT_CONSUME=1, CONSUMED=2};
typedef boost::function1<void, Message&> MessageFunctor;
@@ -126,8 +175,7 @@ class Queue : public boost::enable_share
const std::string name;
MessageStore* store;
const OwnershipToken* owner;
- uint32_t consumerCount; // Actually a count of all subscriptions, acquiring or not.
- uint32_t browserCount; // Count of non-acquiring subscriptions.
+ QueueUsers users;
OwnershipToken* exclusive;
std::vector<std::string> traceExclude;
QueueListeners listeners;
@@ -149,7 +197,6 @@ class Queue : public boost::enable_share
* o Queue::UsageBarrier (TBD: move under separate lock)
*/
mutable qpid::sys::Mutex messageLock;
- mutable qpid::sys::Mutex ownershipLock;
mutable uint64_t persistenceId;
QueueSettings settings;
qpid::framing::FieldTable encodableSettings;
@@ -176,6 +223,9 @@ class Queue : public boost::enable_share
Queue::shared_ptr redirectPeer;
bool redirectSource;
+ bool checkAutoDelete(const qpid::sys::Mutex::ScopedLock&) const;
+ bool isUnused(const qpid::sys::Mutex::ScopedLock&) const;
+ bool isEmpty(const qpid::sys::Mutex::ScopedLock&) const;
virtual void push(Message& msg, bool isRecovery=false);
bool accept(const Message&);
void process(Message& msg);
@@ -191,7 +241,7 @@ class Queue : public boost::enable_share
void observeEnqueue(const Message& msg, const sys::Mutex::ScopedLock& lock);
void observeAcquire(const Message& msg, const sys::Mutex::ScopedLock& lock);
void observeRequeue(const Message& msg, const sys::Mutex::ScopedLock& lock);
- void observeDequeue(const Message& msg, const sys::Mutex::ScopedLock& lock);
+ void observeDequeue(const Message& msg, const sys::Mutex::ScopedLock& lock, ScopedAutoDelete*);
void observeConsumerAdd( const Consumer&, const sys::Mutex::ScopedLock& lock);
void observeConsumerRemove( const Consumer&, const sys::Mutex::ScopedLock& lock);
@@ -203,9 +253,9 @@ class Queue : public boost::enable_share
void abandoned(const Message& message);
bool checkNotDeleted(const Consumer::shared_ptr&);
void notifyDeleted();
- uint32_t remove(uint32_t maxCount, MessagePredicate, MessageFunctor, SubscriptionType);
+ uint32_t remove(uint32_t maxCount, MessagePredicate, MessageFunctor, SubscriptionType, bool triggerAutoDelete);
virtual bool checkDepth(const QueueDepth& increment, const Message&);
-
+ void tryAutoDelete();
public:
typedef std::vector<shared_ptr> vector;
@@ -284,6 +334,14 @@ class Queue : public boost::enable_share
QPID_BROKER_EXTERN void consume(Consumer::shared_ptr c,
bool exclusive = false);
QPID_BROKER_EXTERN void cancel(Consumer::shared_ptr c);
+ /**
+ * Used to indicate that the queue is being used in some other
+ * context than by a subscriber. The controlling flag should only
+ * be set if the mode of use is the one that caused the queue to
+ * be created.
+ */
+ QPID_BROKER_EXTERN void markInUse(bool controlling=false);
+ QPID_BROKER_EXTERN void releaseFromUse(bool controlling=false);
QPID_BROKER_EXTERN uint32_t purge(const uint32_t purge_request=0, //defaults to all messages
boost::shared_ptr<Exchange> dest=boost::shared_ptr<Exchange>(),
@@ -308,6 +366,8 @@ class Queue : public boost::enable_share
inline const qpid::framing::FieldTable& getEncodableSettings() const { return encodableSettings; }
inline bool isAutoDelete() const { return settings.autodelete; }
QPID_BROKER_EXTERN bool canAutoDelete() const;
+ QPID_BROKER_EXTERN void scheduleAutoDelete();
+ QPID_BROKER_EXTERN bool isDeleted() const;
const QueueBindings& getBindings() const { return bindings; }
/**
@@ -340,7 +400,6 @@ class Queue : public boost::enable_share
* exclusive owner
*/
static Queue::shared_ptr restore(QueueRegistry& queues, framing::Buffer& buffer);
- QPID_BROKER_EXTERN static void tryAutoDelete(Broker& broker, Queue::shared_ptr, const std::string& connectionId, const std::string& userId);
virtual void setExternalQueueStore(ExternalQueueStore* inst);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp Wed Jul 10 13:47:05 2013
@@ -58,6 +58,9 @@ const std::string PAGING("qpid.paging");
const std::string MAX_PAGES("qpid.max_pages_loaded");
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_IF_UNUSED_KEY("delete-if-unused");
+const std::string DELETE_IF_UNUSED_AND_EMPTY_KEY("delete-if-unused-and-empty");
const std::string LVQ_LEGACY("qpid.last_value_queue");
const std::string LVQ_LEGACY_KEY("qpid.LVQ_key");
@@ -86,6 +89,7 @@ const QueueSettings::Aliases QueueSettin
QueueSettings::QueueSettings(bool d, bool a) :
durable(d),
autodelete(a),
+ lifetime(DELETE_IF_UNUSED),
isTemporary(false),
priorities(0),
defaultFairshare(0),
@@ -214,6 +218,15 @@ bool QueueSettings::handle(const std::st
} else if (key == FILTER) {
filter = value.asString();
return true;
+ } else if (key == LIFETIME_POLICY) {
+ if (value.asString() == DELETE_IF_UNUSED_KEY) {
+ lifetime = DELETE_IF_UNUSED;
+ } else if (value.asString() == DELETE_IF_UNUSED_AND_EMPTY_KEY) {
+ lifetime = DELETE_IF_UNUSED_AND_EMPTY;
+ } else {
+ QPID_LOG(warning, "Invalid value for " << LIFETIME_POLICY << ": " << value);
+ }
+ return true;
} else {
return false;
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h Wed Jul 10 13:47:05 2013
@@ -40,9 +40,21 @@ namespace broker {
struct QueueSettings
{
QPID_BROKER_EXTERN QueueSettings(bool durable=false, bool autodelete=false);
+ /**
+ * The lifetime policy dictates when an autodelete queue is
+ * eligible for delete.
+ */
+ enum LifetimePolicy
+ {
+ DELETE_IF_UNUSED = 0,
+ DELETE_IF_EMPTY,
+ DELETE_IF_UNUSED_AND_EMPTY,
+ DELETE_ON_CLOSE
+ };
bool durable;
bool autodelete;
+ LifetimePolicy lifetime;
bool isTemporary;
//basic queue types:
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp Wed Jul 10 13:47:05 2013
@@ -449,9 +449,6 @@ void SemanticState::cancel(ConsumerImpl:
Queue::shared_ptr queue = c->getQueue();
if(queue) {
queue->cancel(c);
- // Only run auto-delete for counted consumers.
- if (c->isCounted() && queue->canAutoDelete() && !queue->hasExclusiveOwner())
- Queue::tryAutoDelete(session.getBroker(), queue, connectionId, userID);
}
c->cancel();
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp Wed Jul 10 13:47:05 2013
@@ -227,9 +227,6 @@ void SessionAdapter::QueueHandlerImpl::d
while (!exclusiveQueues.empty()) {
Queue::shared_ptr q(exclusiveQueues.front());
q->releaseExclusiveOwnership();
- if (q->canAutoDelete()) {
- Queue::tryAutoDelete(broker, q, connectionId, userId);
- }
exclusiveQueues.erase(exclusiveQueues.begin());
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp Wed Jul 10 13:47:05 2013
@@ -44,7 +44,7 @@ Connection::Connection(qpid::sys::Output
: BrokerContext(b), ManagedConnection(getBroker(), i),
connection(pn_connection()),
transport(pn_transport()),
- out(o), id(i), haveOutput(true)
+ out(o), id(i), haveOutput(true), closeInitiated(false)
{
if (pn_transport_bind(transport, connection)) {
//error
@@ -130,9 +130,6 @@ size_t Connection::encode(char* buffer,
QPID_LOG_CAT(debug, network, id << " encoded " << n << " bytes from " << size)
haveOutput = true;
return n;
- } else if (n == PN_EOS) {
- haveOutput = size;
- return size;//Is this right?
} else if (n == PN_ERR) {
throw Exception(qpid::amqp::error_conditions::INTERNAL_ERROR, QPID_MSG("Error on output: " << getError()));
} else {
@@ -142,23 +139,29 @@ size_t Connection::encode(char* buffer,
}
bool Connection::canEncode()
{
- try {
- for (Sessions::iterator i = sessions.begin();i != sessions.end(); ++i) {
- if (i->second->dispatch()) haveOutput = true;
+ if (!closeInitiated) {
+ try {
+ for (Sessions::iterator i = sessions.begin();i != sessions.end(); ++i) {
+ if (i->second->dispatch()) haveOutput = true;
+ }
+ process();
+ } catch (const Exception& e) {
+ QPID_LOG(error, id << ": " << e.what());
+ pn_condition_t* error = pn_connection_condition(connection);
+ pn_condition_set_name(error, e.symbol());
+ pn_condition_set_description(error, e.what());
+ close();
+ haveOutput = true;
+ } catch (const std::exception& e) {
+ QPID_LOG(error, id << ": " << e.what());
+ pn_condition_t* error = pn_connection_condition(connection);
+ pn_condition_set_name(error, qpid::amqp::error_conditions::INTERNAL_ERROR.c_str());
+ pn_condition_set_description(error, e.what());
+ close();
+ haveOutput = true;
}
- process();
- } catch (const Exception& e) {
- QPID_LOG(error, id << ": " << e.what());
- pn_condition_t* error = pn_connection_condition(connection);
- pn_condition_set_name(error, e.symbol());
- pn_condition_set_description(error, e.what());
- close();
- } catch (const std::exception& e) {
- QPID_LOG(error, id << ": " << e.what());
- pn_condition_t* error = pn_connection_condition(connection);
- pn_condition_set_name(error, qpid::amqp::error_conditions::INTERNAL_ERROR.c_str());
- pn_condition_set_description(error, e.what());
- close();
+ } else {
+ QPID_LOG(info, "Connection " << id << " has been closed locally");
}
//TODO: proper handling of time in and out of tick
pn_transport_tick(transport, 0);
@@ -195,9 +198,12 @@ void Connection::closed()
}
void Connection::close()
{
- closed();
- QPID_LOG_CAT(debug, model, id << " connection closed");
- pn_connection_close(connection);
+ if (!closeInitiated) {
+ closeInitiated = true;
+ closed();
+ QPID_LOG_CAT(debug, model, id << " connection closed");
+ pn_connection_close(connection);
+ }
}
bool Connection::isClosed() const
{
@@ -256,29 +262,29 @@ void Connection::process()
//handle deliveries
for (pn_delivery_t* delivery = pn_work_head(connection); delivery; delivery = pn_work_next(delivery)) {
pn_link_t* link = pn_delivery_link(delivery);
- if (pn_link_is_receiver(link)) {
- Sessions::iterator i = sessions.find(pn_link_session(link));
- if (i != sessions.end()) {
- try {
+ try {
+ if (pn_link_is_receiver(link)) {
+ Sessions::iterator i = sessions.find(pn_link_session(link));
+ if (i != sessions.end()) {
i->second->readable(link, delivery);
- } catch (const Exception& e) {
- QPID_LOG_CAT(error, protocol, "Error on publish: " << e.what());
- pn_condition_t* error = pn_link_condition(link);
- pn_condition_set_name(error, e.symbol());
- pn_condition_set_description(error, e.what());
- pn_link_close(link);
+ } else {
+ pn_delivery_update(delivery, PN_REJECTED);
+ }
+ } else { //i.e. SENDER
+ Sessions::iterator i = sessions.find(pn_link_session(link));
+ if (i != sessions.end()) {
+ QPID_LOG(trace, id << " handling outgoing delivery for " << link << " on session " << pn_link_session(link));
+ i->second->writable(link, delivery);
+ } else {
+ QPID_LOG(error, id << " Got delivery for non-existent session: " << pn_link_session(link) << ", link: " << link);
}
- } else {
- pn_delivery_update(delivery, PN_REJECTED);
- }
- } else { //i.e. SENDER
- Sessions::iterator i = sessions.find(pn_link_session(link));
- if (i != sessions.end()) {
- QPID_LOG(trace, id << " handling outgoing delivery for " << link << " on session " << pn_link_session(link));
- i->second->writable(link, delivery);
- } else {
- QPID_LOG(error, id << " Got delivery for non-existent session: " << pn_link_session(link) << ", link: " << link);
}
+ } catch (const Exception& e) {
+ QPID_LOG_CAT(error, protocol, "Error processing deliveries: " << e.what());
+ pn_condition_t* error = pn_link_condition(link);
+ pn_condition_set_name(error, e.symbol());
+ pn_condition_set_description(error, e.what());
+ pn_link_close(link);
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.h?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.h Wed Jul 10 13:47:05 2013
@@ -67,6 +67,7 @@ class Connection : public BrokerContext,
const std::string id;
bool haveOutput;
Sessions sessions;
+ bool closeInitiated;
virtual void process();
std::string getError();
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/DataReader.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/DataReader.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/DataReader.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/DataReader.cpp Wed Jul 10 13:47:05 2013
@@ -160,13 +160,15 @@ void DataReader::readArray(pn_data_t* /*
void DataReader::readList(pn_data_t* data, const qpid::amqp::Descriptor* descriptor)
{
size_t count = pn_data_get_list(data);
- reader.onStartList(count, qpid::amqp::CharSequence(), descriptor);
- pn_data_enter(data);
- for (size_t i = 0; i < count && pn_data_next(data); ++i) {
- read(data);
+ bool skip = reader.onStartList(count, qpid::amqp::CharSequence(), descriptor);
+ if (!skip) {
+ pn_data_enter(data);
+ for (size_t i = 0; i < count && pn_data_next(data); ++i) {
+ read(data);
+ }
+ pn_data_exit(data);
+ reader.onEndList(count, descriptor);
}
- pn_data_exit(data);
- reader.onEndList(count, descriptor);
}
void DataReader::readMap(pn_data_t* data, const qpid::amqp::Descriptor* descriptor)
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.cpp Wed Jul 10 13:47:05 2013
@@ -20,10 +20,16 @@
*/
#include "qpid/broker/amqp/NodeProperties.h"
#include "qpid/broker/amqp/DataReader.h"
+#include "qpid/broker/QueueSettings.h"
#include "qpid/amqp/CharSequence.h"
+#include "qpid/amqp/Descriptor.h"
+#include "qpid/amqp/descriptors.h"
#include "qpid/types/Variant.h"
#include "qpid/broker/QueueSettings.h"
#include "qpid/log/Statement.h"
+extern "C" {
+#include <proton/engine.h>
+}
using qpid::amqp::CharSequence;
using qpid::amqp::Descriptor;
@@ -36,6 +42,7 @@ namespace {
const std::string MOVE("move");
const std::string COPY("copy");
const std::string SUPPORTED_DIST_MODES("supported-dist-modes");
+const std::string LIFETIME_POLICY("lifetime-policy");
//AMQP 0-10 standard parameters:
const std::string DURABLE("durable");
@@ -43,9 +50,57 @@ const std::string EXCLUSIVE("exclusive")
const std::string AUTO_DELETE("auto-delete");
const std::string ALTERNATE_EXCHANGE("alternate-exchange");
const std::string EXCHANGE_TYPE("exchange-type");
+
+pn_bytes_t convert(const std::string& s)
+{
+ pn_bytes_t result;
+ result.start = const_cast<char*>(s.data());
+ result.size = s.size();
+ return result;
+}
+
+bool getLifetimePolicy(const Descriptor& d, QueueSettings::LifetimePolicy& policy)
+{
+ if (d.match(qpid::amqp::lifetime_policy::DELETE_ON_CLOSE_SYMBOL, qpid::amqp::lifetime_policy::DELETE_ON_CLOSE_CODE)) {
+ policy = QueueSettings::DELETE_ON_CLOSE;
+ return true;
+ } else if (d.match(qpid::amqp::lifetime_policy::DELETE_ON_NO_LINKS_SYMBOL, qpid::amqp::lifetime_policy::DELETE_ON_NO_LINKS_CODE)) {
+ policy = QueueSettings::DELETE_IF_UNUSED;
+ return true;
+ } else if (d.match(qpid::amqp::lifetime_policy::DELETE_ON_NO_MESSAGES_SYMBOL, qpid::amqp::lifetime_policy::DELETE_ON_NO_MESSAGES_CODE)) {
+ policy = QueueSettings::DELETE_IF_EMPTY;
+ return true;
+ } else if (d.match(qpid::amqp::lifetime_policy::DELETE_ON_NO_LINKS_OR_MESSAGES_SYMBOL, qpid::amqp::lifetime_policy::DELETE_ON_NO_LINKS_OR_MESSAGES_CODE)) {
+ policy = QueueSettings::DELETE_IF_UNUSED_AND_EMPTY;
+ return true;
+ } else {
+ return false;
+ }
+}
+
+bool getLifetimeDescriptorSymbol(QueueSettings::LifetimePolicy policy, pn_bytes_t& out)
+{
+ switch (policy) {
+ case QueueSettings::DELETE_ON_CLOSE:
+ out = convert(qpid::amqp::lifetime_policy::DELETE_ON_CLOSE_SYMBOL);
+ return true;
+ case QueueSettings::DELETE_IF_UNUSED:
+ out = convert(qpid::amqp::lifetime_policy::DELETE_ON_NO_LINKS_SYMBOL);
+ return true;
+ case QueueSettings::DELETE_IF_EMPTY:
+ out = convert(qpid::amqp::lifetime_policy::DELETE_ON_NO_MESSAGES_SYMBOL);
+ return true;
+ case QueueSettings::DELETE_IF_UNUSED_AND_EMPTY:
+ out = convert(qpid::amqp::lifetime_policy::DELETE_ON_NO_LINKS_OR_MESSAGES_SYMBOL);
+ return true;
+ default:
+ return false;
+ }
+}
+
}
-NodeProperties::NodeProperties() : queue(true), durable(false), autoDelete(false), exclusive(false), exchangeType("topic") {}
+NodeProperties::NodeProperties() : queue(true), durable(false), autoDelete(false), exclusive(false), exchangeType("topic"), lifetime(QueueSettings::DELETE_IF_UNUSED) {}
void NodeProperties::read(pn_data_t* data)
{
@@ -53,12 +108,38 @@ void NodeProperties::read(pn_data_t* dat
reader.read(data);
}
-void NodeProperties::process(const std::string& key, const qpid::types::Variant& value)
+void NodeProperties::write(pn_data_t* data)
+{
+ pn_data_put_map(data);
+ pn_data_enter(data);
+ pn_data_put_symbol(data, convert(SUPPORTED_DIST_MODES));
+ pn_data_put_string(data, convert(queue ? MOVE : COPY));
+ pn_bytes_t symbol;
+ if (autoDelete && getLifetimeDescriptorSymbol(lifetime, symbol)) {
+ pn_data_put_symbol(data, convert(LIFETIME_POLICY));
+ pn_data_put_described(data);
+ pn_data_enter(data);
+ pn_data_put_symbol(data, symbol);
+ pn_data_put_list(data);
+ pn_data_exit(data);
+ }
+ pn_data_exit(data);
+}
+
+void NodeProperties::process(const std::string& key, const qpid::types::Variant& value, const Descriptor* d)
{
- QPID_LOG(notice, "Processing node property " << key << " = " << value);
+ QPID_LOG(debug, "Processing node property " << key << " = " << value);
if (key == SUPPORTED_DIST_MODES) {
if (value == MOVE) queue = true;
else if (value == COPY) queue = false;
+ } else if (key == LIFETIME_POLICY) {
+ if (d) {
+ if (getLifetimePolicy(*d, lifetime)) {
+ autoDelete = true;
+ } else {
+ QPID_LOG(warning, "Unrecognised lifetime policy: " << *d);
+ }
+ }
} else if (key == DURABLE) {
durable = value;
} else if (key == EXCLUSIVE) {
@@ -74,84 +155,91 @@ void NodeProperties::process(const std::
}
}
-void NodeProperties::onNullValue(const CharSequence& key, const Descriptor*)
+bool NodeProperties::onStartListValue(const qpid::amqp::CharSequence& key, uint32_t count, const qpid::amqp::Descriptor* d)
{
- process(key.str(), qpid::types::Variant());
+ QPID_LOG(debug, "NodeProperties::onStartListValue(" << std::string(key.data, key.size) << ", " << count << ", " << d);
+ process(key.str(), qpid::types::Variant(), d);
+ return true;
}
-void NodeProperties::onBooleanValue(const CharSequence& key, bool value, const Descriptor*)
+void NodeProperties::onNullValue(const CharSequence& key, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), qpid::types::Variant(), d);
}
-void NodeProperties::onUByteValue(const CharSequence& key, uint8_t value, const Descriptor*)
+void NodeProperties::onBooleanValue(const CharSequence& key, bool value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value, d);
}
-void NodeProperties::onUShortValue(const CharSequence& key, uint16_t value, const Descriptor*)
+void NodeProperties::onUByteValue(const CharSequence& key, uint8_t value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value, d);
}
-void NodeProperties::onUIntValue(const CharSequence& key, uint32_t value, const Descriptor*)
+void NodeProperties::onUShortValue(const CharSequence& key, uint16_t value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value, d);
}
-void NodeProperties::onULongValue(const CharSequence& key, uint64_t value, const Descriptor*)
+void NodeProperties::onUIntValue(const CharSequence& key, uint32_t value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value, d);
}
-void NodeProperties::onByteValue(const CharSequence& key, int8_t value, const Descriptor*)
+void NodeProperties::onULongValue(const CharSequence& key, uint64_t value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value, d);
}
-void NodeProperties::onShortValue(const CharSequence& key, int16_t value, const Descriptor*)
+void NodeProperties::onByteValue(const CharSequence& key, int8_t value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value, d);
}
-void NodeProperties::onIntValue(const CharSequence& key, int32_t value, const Descriptor*)
+void NodeProperties::onShortValue(const CharSequence& key, int16_t value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value, d);
}
-void NodeProperties::onLongValue(const CharSequence& key, int64_t value, const Descriptor*)
+void NodeProperties::onIntValue(const CharSequence& key, int32_t value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value, d);
}
-void NodeProperties::onFloatValue(const CharSequence& key, float value, const Descriptor*)
+void NodeProperties::onLongValue(const CharSequence& key, int64_t value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value, d);
}
-void NodeProperties::onDoubleValue(const CharSequence& key, double value, const Descriptor*)
+void NodeProperties::onFloatValue(const CharSequence& key, float value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value, d);
}
-void NodeProperties::onUuidValue(const CharSequence& key, const CharSequence& value, const Descriptor*)
+void NodeProperties::onDoubleValue(const CharSequence& key, double value, const Descriptor* d)
{
- process(key.str(), value.str());
+ process(key.str(), value, d);
}
-void NodeProperties::onTimestampValue(const CharSequence& key, int64_t value, const Descriptor*)
+void NodeProperties::onUuidValue(const CharSequence& key, const CharSequence& value, const Descriptor* d)
{
- process(key.str(), value);
+ process(key.str(), value.str(), d);
}
-void NodeProperties::onStringValue(const CharSequence& key, const CharSequence& value, const Descriptor*)
+void NodeProperties::onTimestampValue(const CharSequence& key, int64_t value, const Descriptor* d)
{
- process(key.str(), value.str());
+ process(key.str(), value, d);
}
-void NodeProperties::onSymbolValue(const CharSequence& key, const CharSequence& value, const Descriptor*)
+void NodeProperties::onStringValue(const CharSequence& key, const CharSequence& value, const Descriptor* d)
{
- process(key.str(), value.str());
+ process(key.str(), value.str(), d);
+}
+
+void NodeProperties::onSymbolValue(const CharSequence& key, const CharSequence& value, const Descriptor* d)
+{
+ process(key.str(), value.str(), d);
}
QueueSettings NodeProperties::getQueueSettings()
@@ -159,6 +247,7 @@ QueueSettings NodeProperties::getQueueSe
QueueSettings settings(durable, autoDelete);
qpid::types::Variant::Map unused;
settings.populate(properties, unused);
+ settings.lifetime = lifetime;
return settings;
}
@@ -183,4 +272,9 @@ std::string NodeProperties::getAlternate
return alternateExchange;
}
+bool NodeProperties::trackControllingLink() const
+{
+ return lifetime == QueueSettings::DELETE_ON_CLOSE || lifetime == QueueSettings::DELETE_IF_EMPTY;
+}
+
}}} // namespace qpid::broker::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.h?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.h Wed Jul 10 13:47:05 2013
@@ -23,6 +23,7 @@
*/
#include "qpid/amqp/MapReader.h"
#include "qpid/types/Variant.h"
+#include "qpid/broker/QueueSettings.h"
struct pn_data_t;
namespace qpid {
@@ -35,6 +36,7 @@ class NodeProperties : public qpid::amqp
public:
NodeProperties();
void read(pn_data_t*);
+ void write(pn_data_t*);
void onNullValue(const qpid::amqp::CharSequence&, const qpid::amqp::Descriptor*);
void onBooleanValue(const qpid::amqp::CharSequence&, bool, const qpid::amqp::Descriptor*);
void onUByteValue(const qpid::amqp::CharSequence&, uint8_t, const qpid::amqp::Descriptor*);
@@ -51,12 +53,14 @@ class NodeProperties : public qpid::amqp
void onTimestampValue(const qpid::amqp::CharSequence&, int64_t, const qpid::amqp::Descriptor*);
void onStringValue(const qpid::amqp::CharSequence&, const qpid::amqp::CharSequence&, const qpid::amqp::Descriptor*);
void onSymbolValue(const qpid::amqp::CharSequence&, const qpid::amqp::CharSequence&, const qpid::amqp::Descriptor*);
+ bool onStartListValue(const qpid::amqp::CharSequence&, uint32_t count, const qpid::amqp::Descriptor*);
bool isQueue() const;
QueueSettings getQueueSettings();
bool isDurable() const;
bool isExclusive() const;
std::string getExchangeType() const;
std::string getAlternateExchange() const;
+ bool trackControllingLink() const;
private:
bool queue;
bool durable;
@@ -65,8 +69,9 @@ class NodeProperties : public qpid::amqp
std::string exchangeType;
std::string alternateExchange;
qpid::types::Variant::Map properties;
+ QueueSettings::LifetimePolicy lifetime;
- void process(const std::string&, const qpid::types::Variant&);
+ void process(const std::string&, const qpid::types::Variant&, const qpid::amqp::Descriptor*);
};
}}} // namespace qpid::broker::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp Wed Jul 10 13:47:05 2013
@@ -19,6 +19,7 @@
*
*/
#include "qpid/broker/amqp/Outgoing.h"
+#include "qpid/broker/amqp/Exception.h"
#include "qpid/broker/amqp/Header.h"
#include "qpid/broker/amqp/Session.h"
#include "qpid/broker/amqp/Translation.h"
@@ -26,7 +27,9 @@
#include "qpid/broker/Selector.h"
#include "qpid/broker/TopicKeyNode.h"
#include "qpid/sys/OutputControl.h"
+#include "qpid/amqp/descriptors.h"
#include "qpid/amqp/MessageEncoder.h"
+#include "qpid/framing/reply_exceptions.h"
#include "qpid/log/Statement.h"
namespace qpid {
@@ -41,10 +44,11 @@ void Outgoing::wakeup()
session.wakeup();
}
-OutgoingFromQueue::OutgoingFromQueue(Broker& broker, const std::string& source, const std::string& target, boost::shared_ptr<Queue> q, pn_link_t* l, Session& session, qpid::sys::OutputControl& o, bool e)
+OutgoingFromQueue::OutgoingFromQueue(Broker& broker, const std::string& source, const std::string& target, boost::shared_ptr<Queue> q, pn_link_t* l, Session& session, qpid::sys::OutputControl& o, bool e, bool p)
: Outgoing(broker, session, source, target, pn_link_name(l)),
Consumer(pn_link_name(l), /*FIXME*/CONSUMER),
exclusive(e),
+ isControllingUser(p),
queue(q), deliveries(5000), link(l), out(o),
current(0), outstanding(0),
buffer(1024)/*used only for header at present*/
@@ -52,6 +56,7 @@ OutgoingFromQueue::OutgoingFromQueue(Bro
for (size_t i = 0 ; i < deliveries.capacity(); ++i) {
deliveries[i].init(i);
}
+ if (isControllingUser) queue->markInUse(true);
}
void OutgoingFromQueue::init()
@@ -63,11 +68,15 @@ bool OutgoingFromQueue::doWork()
{
QPID_LOG(trace, "Dispatching to " << getName() << ": " << pn_link_credit(link));
if (canDeliver()) {
- if (queue->dispatch(shared_from_this())) {
- return true;
- } else {
- pn_link_drained(link);
- QPID_LOG(debug, "No message available on " << queue->getName());
+ try{
+ if (queue->dispatch(shared_from_this())) {
+ return true;
+ } else {
+ pn_link_drained(link);
+ QPID_LOG(debug, "No message available on " << queue->getName());
+ }
+ } catch (const qpid::framing::ResourceDeletedException& e) {
+ throw Exception(qpid::amqp::error_conditions::RESOURCE_DELETED, e.what());
}
} else {
QPID_LOG(debug, "Can't deliver to " << getName() << " from " << queue->getName() << ": " << pn_link_credit(link));
@@ -142,14 +151,14 @@ bool OutgoingFromQueue::canDeliver()
void OutgoingFromQueue::detached()
{
- QPID_LOG(debug, "Detaching outgoing link from " << queue->getName());
+ QPID_LOG(debug, "Detaching outgoing link " << getName() << " from " << queue->getName());
queue->cancel(shared_from_this());
//TODO: release in a clearer order?
for (size_t i = 0 ; i < deliveries.capacity(); ++i) {
if (deliveries[i].msg) queue->release(deliveries[i].cursor, true);
}
if (exclusive) queue->releaseExclusiveOwnership();
- Queue::tryAutoDelete(*queue->getBroker(), queue, "", "");
+ else if (isControllingUser) queue->releaseFromUse(true);
}
//Consumer interface:
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.h?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.h Wed Jul 10 13:47:05 2013
@@ -88,7 +88,7 @@ class Outgoing : public ManagedOutgoingL
class OutgoingFromQueue : public Outgoing, public qpid::broker::Consumer, public boost::enable_shared_from_this<OutgoingFromQueue>
{
public:
- OutgoingFromQueue(Broker&, const std::string& source, const std::string& target, boost::shared_ptr<Queue> q, pn_link_t* l, Session&, qpid::sys::OutputControl& o, bool exclusive);
+ OutgoingFromQueue(Broker&, const std::string& source, const std::string& target, boost::shared_ptr<Queue> q, pn_link_t* l, Session&, qpid::sys::OutputControl& o, bool exclusive, bool isControllingUser);
void setSubjectFilter(const std::string&);
void setSelectorFilter(const std::string&);
void init();
@@ -124,6 +124,7 @@ class OutgoingFromQueue : public Outgoin
};
const bool exclusive;
+ const bool isControllingUser;
boost::shared_ptr<Queue> queue;
CircularArray<Record> deliveries;
pn_link_t* link;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp Wed Jul 10 13:47:05 2013
@@ -24,9 +24,11 @@
#include "Message.h"
#include "Connection.h"
#include "Domain.h"
+#include "Exception.h"
#include "Interconnects.h"
#include "Relay.h"
#include "Topic.h"
+#include "qpid/amqp/descriptors.h"
#include "qpid/broker/Broker.h"
#include "qpid/broker/DeliverableMessage.h"
#include "qpid/broker/Exchange.h"
@@ -112,10 +114,16 @@ void setCapabilities(pn_data_t* in, pn_d
class IncomingToQueue : public DecodingIncoming
{
public:
- IncomingToQueue(Broker& b, Session& p, boost::shared_ptr<qpid::broker::Queue> q, pn_link_t* l, const std::string& source) : DecodingIncoming(l, b, p, source, q->getName(), pn_link_name(l)), queue(q) {}
+ IncomingToQueue(Broker& b, Session& p, boost::shared_ptr<qpid::broker::Queue> q, pn_link_t* l, const std::string& source, bool icl)
+ : DecodingIncoming(l, b, p, source, q->getName(), pn_link_name(l)), queue(q), isControllingLink(icl)
+ {
+ queue->markInUse(isControllingLink);
+ }
+ ~IncomingToQueue() { queue->releaseFromUse(isControllingLink); }
void handle(qpid::broker::Message& m);
private:
boost::shared_ptr<qpid::broker::Queue> queue;
+ bool isControllingLink;
};
class IncomingToExchange : public DecodingIncoming
@@ -152,6 +160,7 @@ Session::ResolvedNode Session::resolve(c
node.exchange = connection.getBroker().createExchange(name, node.properties.getExchangeType(), node.properties.isDurable(), node.properties.getAlternateExchange(),
args, connection.getUserId(), connection.getId()).first;
}
+ node.created = true;
} else {
size_t i = name.find('@');
if (i != std::string::npos && (i+1) < name.length()) {
@@ -188,7 +197,12 @@ Session::ResolvedNode Session::resolve(c
std::string Session::generateName(pn_link_t* link)
{
std::stringstream s;
- s << qpid::types::Uuid(true) << "::" << pn_link_name(link);
+ if (connection.getContainerId().empty()) {
+ s << qpid::types::Uuid(true);
+ } else {
+ s << connection.getContainerId();
+ }
+ s << "_" << pn_link_name(link);
return s.str();
}
@@ -210,7 +224,7 @@ void Session::attach(pn_link_t* link)
//i.e a subscription
std::string name;
if (pn_terminus_get_type(source) == PN_UNSPECIFIED) {
- throw qpid::Exception("No source specified!");/*invalid-field?*/
+ throw Exception(qpid::amqp::error_conditions::PRECONDITION_FAILED, "No source specified!");
} else if (pn_terminus_is_dynamic(source)) {
name = generateName(link);
QPID_LOG(debug, "Received attach request for outgoing link from " << name);
@@ -226,7 +240,7 @@ void Session::attach(pn_link_t* link)
pn_terminus_t* target = pn_link_remote_target(link);
std::string name;
if (pn_terminus_get_type(target) == PN_UNSPECIFIED) {
- throw qpid::Exception("No target specified!");/*invalid field?*/
+ throw Exception(qpid::amqp::error_conditions::PRECONDITION_FAILED, "No target specified!");
} else if (pn_terminus_is_dynamic(target)) {
name = generateName(link);
QPID_LOG(debug, "Received attach request for incoming link to " << name);
@@ -252,6 +266,9 @@ void Session::setupIncoming(pn_link_t* l
setCapabilities(pn_terminus_capabilities(target), pn_terminus_capabilities(pn_link_target(link)), node.exchange);
authorise.incoming(node.exchange);
}
+ if (node.created) {
+ node.properties.write(pn_terminus_properties(pn_link_target(link)));
+ }
const char* sourceAddress = pn_terminus_get_address(pn_link_remote_source(link));
if (!sourceAddress) {
@@ -262,7 +279,7 @@ void Session::setupIncoming(pn_link_t* l
source = sourceAddress;
}
if (node.queue) {
- boost::shared_ptr<Incoming> q(new IncomingToQueue(connection.getBroker(), *this, node.queue, link, source));
+ boost::shared_ptr<Incoming> q(new IncomingToQueue(connection.getBroker(), *this, node.queue, link, source, node.properties.trackControllingLink()));
incoming[link] = q;
} else if (node.exchange) {
boost::shared_ptr<Incoming> e(new IncomingToExchange(connection.getBroker(), *this, node.exchange, link, source));
@@ -272,7 +289,7 @@ void Session::setupIncoming(pn_link_t* l
incoming[link] = in;
} else {
pn_terminus_set_type(pn_link_target(link), PN_UNSPECIFIED);
- throw qpid::Exception("Node not found: " + name);/*not-found*/
+ throw Exception(qpid::amqp::error_conditions::NOT_FOUND, std::string("Node not found: ") + name);
}
if (connection.getBroker().getOptions().auth && !connection.isLink())
incoming[link]->verify(connection.getUserId(), connection.getBroker().getOptions().realm);
@@ -284,6 +301,9 @@ void Session::setupOutgoing(pn_link_t* l
ResolvedNode node = resolve(name, source, false);
if (node.queue) setCapabilities(pn_terminus_capabilities(source), pn_terminus_capabilities(pn_link_source(link)), node.queue);
else if (node.exchange) setCapabilities(pn_terminus_capabilities(source), pn_terminus_capabilities(pn_link_source(link)), node.exchange);
+ if (node.created) {
+ node.properties.write(pn_terminus_properties(pn_link_source(link)));
+ }
Filter filter;
filter.read(pn_terminus_filter(source));
@@ -299,7 +319,7 @@ void Session::setupOutgoing(pn_link_t* l
if (node.queue) {
authorise.outgoing(node.queue);
- boost::shared_ptr<Outgoing> q(new OutgoingFromQueue(connection.getBroker(), name, target, node.queue, link, *this, out, false));
+ boost::shared_ptr<Outgoing> q(new OutgoingFromQueue(connection.getBroker(), name, target, node.queue, link, *this, out, false, node.properties.trackControllingLink()));
q->init();
filter.apply(q);
outgoing[link] = q;
@@ -330,7 +350,7 @@ void Session::setupOutgoing(pn_link_t* l
if (!shared) queue->setExclusiveOwner(this);
authorise.outgoing(node.exchange, queue, filter);
filter.bind(node.exchange, queue);
- boost::shared_ptr<Outgoing> q(new OutgoingFromQueue(connection.getBroker(), name, target, queue, link, *this, out, !shared));
+ boost::shared_ptr<Outgoing> q(new OutgoingFromQueue(connection.getBroker(), name, target, queue, link, *this, out, !shared, false));
outgoing[link] = q;
q->init();
} else if (node.relay) {
@@ -339,7 +359,7 @@ void Session::setupOutgoing(pn_link_t* l
out->init();
} else {
pn_terminus_set_type(pn_link_source(link), PN_UNSPECIFIED);
- throw qpid::Exception("Node not found: " + name);/*not-found*/
+ throw Exception(qpid::amqp::error_conditions::NOT_FOUND, std::string("Node not found: ") + name);/*not-found*/
}
filter.write(pn_terminus_filter(pn_link_source(link)));
QPID_LOG(debug, "Outgoing link attached");
@@ -438,8 +458,19 @@ void Session::writable(pn_link_t* link,
bool Session::dispatch()
{
bool output(false);
- for (OutgoingLinks::iterator s = outgoing.begin(); s != outgoing.end(); ++s) {
- if (s->second->doWork()) output = true;
+ for (OutgoingLinks::iterator s = outgoing.begin(); s != outgoing.end();) {
+ try {
+ if (s->second->doWork()) output = true;
+ ++s;
+ } catch (const Exception& e) {
+ pn_condition_t* error = pn_link_condition(s->first);
+ pn_condition_set_name(error, e.symbol());
+ pn_condition_set_description(error, e.what());
+ pn_link_close(s->first);
+ s->second->detached();
+ outgoing.erase(s++);
+ output = true;
+ }
}
if (completed.size()) {
output = true;
@@ -452,8 +483,19 @@ bool Session::dispatch()
accepted(*i, true);
}
}
- for (IncomingLinks::iterator i = incoming.begin(); i != incoming.end(); ++i) {
- if (i->second->doWork()) output = true;
+ for (IncomingLinks::iterator i = incoming.begin(); i != incoming.end();) {
+ try {
+ if (i->second->doWork()) output = true;
+ ++i;
+ } catch (const Exception& e) {
+ pn_condition_t* error = pn_link_condition(i->first);
+ pn_condition_set_name(error, e.symbol());
+ pn_condition_set_description(error, e.what());
+ pn_link_close(i->first);
+ i->second->detached();
+ incoming.erase(i++);
+ output = true;
+ }
}
return output;
@@ -470,7 +512,7 @@ void Session::close()
}
outgoing.clear();
incoming.clear();
- QPID_LOG(debug, "Session closed, all links detached.");
+ QPID_LOG(debug, "Session " << session << " closed, all links detached.");
for (std::set< boost::shared_ptr<Queue> >::const_iterator i = exclusiveQueues.begin(); i != exclusiveQueues.end(); ++i) {
(*i)->releaseExclusiveOwnership();
}
@@ -490,6 +532,11 @@ Authorise& Session::getAuthorise()
void IncomingToQueue::handle(qpid::broker::Message& message)
{
+ if (queue->isDeleted()) {
+ std::stringstream msg;
+ msg << " Queue " << queue->getName() << " has been deleted";
+ throw Exception(qpid::amqp::error_conditions::RESOURCE_DELETED, msg.str());
+ }
queue->deliver(message);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h Wed Jul 10 13:47:05 2013
@@ -100,6 +100,8 @@ class Session : public ManagedSession, p
boost::shared_ptr<qpid::broker::amqp::Topic> topic;
boost::shared_ptr<Relay> relay;
NodeProperties properties;
+ bool created;
+ ResolvedNode() : created(false) {}
};
ResolvedNode resolve(const std::string name, pn_terminus_t* terminus, bool incoming);
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp Wed Jul 10 13:47:05 2013
@@ -871,7 +871,8 @@ void BrokerReplicator::autoDeleteCheck(b
if (qr->getQueue()->isAutoDelete() && qr->isSubscribed()) {
if (qr->getQueue()->getSettings().autoDeleteDelay) {
// Start the auto-delete timer
- Queue::tryAutoDelete(broker, qr->getQueue(), remoteHost, userId);
+ qr->getQueue()->releaseFromUse();
+ qr->getQueue()->scheduleAutoDelete();
}
else {
// Delete immediately. Don't purge, the primary is gone so we need
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp Wed Jul 10 13:47:05 2013
@@ -121,6 +121,7 @@ QueueReplicator::QueueReplicator(HaBroke
framing::FieldTable args = getArgs();
args.setString(QPID_REPLICATE, printable(NONE).str());
setArgs(args);
+ if (q->isAutoDelete()) q->markInUse();
}
// This must be called immediately after the constructor.
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp Wed Jul 10 13:47:05 2013
@@ -75,6 +75,12 @@ const std::string MOVE("move");
const std::string COPY("copy");
const std::string SUPPORTED_DIST_MODES("supported-dist-modes");
+const std::string AUTO_DELETE("auto-delete");
+const std::string LIFETIME_POLICY("lifetime-policy");
+const std::string DELETE_ON_CLOSE("delete-on-close");
+const std::string DELETE_IF_UNUSED("delete-if-unused");
+const std::string DELETE_IF_EMPTY("delete-if-empty");
+const std::string DELETE_IF_UNUSED_AND_EMPTY("delete-if-unused-and-empty");
const std::string CREATE_ON_DEMAND("create-on-demand");
const std::string DUMMY(".");
@@ -308,6 +314,10 @@ AddressHelper::AddressHelper(const Addre
add(properties, x_declare);
node.erase(i);
}
+ //for temp queues, if neither lifetime-policy nor autodelete are specified, assume delete-on-close
+ if (isTemporary && properties.find(LIFETIME_POLICY) == properties.end() && properties.find(AUTO_DELETE) == properties.end()) {
+ properties[LIFETIME_POLICY] = DELETE_ON_CLOSE;
+ }
if (properties.size() && !(isTemporary || createPolicy.size())) {
QPID_LOG(warning, "Properties will be ignored! " << address);
@@ -559,7 +569,24 @@ std::string AddressHelper::getLinkName(c
return name.str();
}
}
-
+namespace {
+std::string toLifetimePolicy(const std::string& value)
+{
+ if (value == DELETE_ON_CLOSE) return qpid::amqp::lifetime_policy::DELETE_ON_CLOSE_SYMBOL;
+ else if (value == DELETE_IF_UNUSED) return qpid::amqp::lifetime_policy::DELETE_ON_NO_LINKS_SYMBOL;
+ else if (value == DELETE_IF_EMPTY) return qpid::amqp::lifetime_policy::DELETE_ON_NO_MESSAGES_SYMBOL;
+ else if (value == DELETE_IF_UNUSED_AND_EMPTY) return qpid::amqp::lifetime_policy::DELETE_ON_NO_LINKS_OR_MESSAGES_SYMBOL;
+ else return value;//asume value is itself the symbolic descriptor
+}
+void putLifetimePolicy(pn_data_t* data, const std::string& value)
+{
+ pn_data_put_described(data);
+ pn_data_enter(data);
+ pn_data_put_symbol(data, convert(value));
+ pn_data_put_list(data);
+ pn_data_exit(data);
+}
+}
void AddressHelper::setNodeProperties(pn_terminus_t* terminus)
{
if (properties.size() || type.size()) {
@@ -575,8 +602,13 @@ void AddressHelper::setNodeProperties(pn
pn_data_put_bool(data, true);
}
for (qpid::types::Variant::Map::const_iterator i = properties.begin(); i != properties.end(); ++i) {
- pn_data_put_symbol(data, convert(i->first));
- pn_data_put_string(data, convert(i->second.asString()));
+ if (i->first == LIFETIME_POLICY) {
+ pn_data_put_symbol(data, convert(i->first));
+ putLifetimePolicy(data, toLifetimePolicy(i->second.asString()));
+ } else {
+ pn_data_put_symbol(data, convert(i->first));
+ pn_data_put_string(data, convert(i->second.asString()));
+ }
}
pn_data_exit(data);
}
Modified: qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp?rev=1501768&r1=1501767&r2=1501768&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp Wed Jul 10 13:47:05 2013
@@ -388,6 +388,7 @@ QPID_AUTO_TEST_CASE(testGroupsMultiConsu
// Owners= ---, ---, ---
TestConsumer::shared_ptr c3(new TestConsumer("C3"));
+ queue->consume(c3);
std::deque<QueueCursor> dequeMeC3;
verifyAcquire(queue, c3, dequeMeC3, "a", 2 );
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org