diff options
| -rw-r--r-- | cpp/src/qpid/broker/PersistableMessage.cpp | 2 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/PersistableQueue.h | 1 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/Queue.cpp | 10 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/Queue.h | 2 |
4 files changed, 13 insertions, 2 deletions
diff --git a/cpp/src/qpid/broker/PersistableMessage.cpp b/cpp/src/qpid/broker/PersistableMessage.cpp index 62396ad995..21fb421168 100644 --- a/cpp/src/qpid/broker/PersistableMessage.cpp +++ b/cpp/src/qpid/broker/PersistableMessage.cpp @@ -52,7 +52,7 @@ void PersistableMessage::flush() } for (syncList::iterator i = copy.begin(); i != copy.end(); ++i) { PersistableQueue::shared_ptr q(i->lock()); - if (q) { + if (q && q->isValid()) { store->flush(*q); } } diff --git a/cpp/src/qpid/broker/PersistableQueue.h b/cpp/src/qpid/broker/PersistableQueue.h index e742a72f42..bcf4268158 100644 --- a/cpp/src/qpid/broker/PersistableQueue.h +++ b/cpp/src/qpid/broker/PersistableQueue.h @@ -60,6 +60,7 @@ public: }; virtual void setExternalQueueStore(ExternalQueueStore* inst) = 0; + virtual bool isValid() = 0; inline ExternalQueueStore* getExternalQueueStore() const {return externalQueueStore;}; diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp index 417aaddb4a..90b0d4cc52 100644 --- a/cpp/src/qpid/broker/Queue.cpp +++ b/cpp/src/qpid/broker/Queue.cpp @@ -725,6 +725,7 @@ bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg, Messages dequeues; { Mutex::ScopedLock locker(messageLock); + if (deleted) return false; policy->tryEnqueue(msg); policy->getPendingDequeues(dequeues); } @@ -766,7 +767,7 @@ bool Queue::dequeue(TransactionContext* ctxt, const QueuedMessage& msg) { { Mutex::ScopedLock locker(messageLock); - if (!isEnqueued(msg)) return false; + if (deleted || !isEnqueued(msg)) return false; if (!ctxt) { dequeued(msg); } @@ -892,6 +893,7 @@ void Queue::destroy() popAndDequeue(); } alternateExchange->decAlternateUsers(); + deleted = true; } if (store) { @@ -1169,3 +1171,9 @@ void Queue::checkNotDeleted() throw ResourceDeletedException(QPID_MSG("Queue " << getName() << " has been deleted.")); } } + +bool Queue::isValid() +{ + Mutex::ScopedLock locker(messageLock); + return !deleted; +} diff --git a/cpp/src/qpid/broker/Queue.h b/cpp/src/qpid/broker/Queue.h index 0b690e7cc5..51b64ba036 100644 --- a/cpp/src/qpid/broker/Queue.h +++ b/cpp/src/qpid/broker/Queue.h @@ -341,6 +341,8 @@ namespace qpid { * has been recovered in the prepared state (dtx only) */ void recoverPrepared(boost::intrusive_ptr<Message>& msg); + + bool isValid(); }; } } |