QPID-2588: Ensure we do not make calls on store for a given queue once it has been destroyed

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@944016 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 13 insertions, 2 deletions

diff --git a/qpid/cpp/src/qpid/broker/PersistableMessage.cpp b/qpid/cpp/src/qpid/broker/PersistableMessage.cpp
index 62396ad995..21fb421168 100644
--- a/qpid/cpp/src/qpid/broker/PersistableMessage.cpp
+++ b/qpid/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/qpid/cpp/src/qpid/broker/PersistableQueue.h b/qpid/cpp/src/qpid/broker/PersistableQueue.h
index e742a72f42..bcf4268158 100644
--- a/qpid/cpp/src/qpid/broker/PersistableQueue.h
+++ b/qpid/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/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp
index 417aaddb4a..90b0d4cc52 100644
--- a/qpid/cpp/src/qpid/broker/Queue.cpp
+++ b/qpid/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/qpid/cpp/src/qpid/broker/Queue.h b/qpid/cpp/src/qpid/broker/Queue.h
index 0b690e7cc5..51b64ba036 100644
--- a/qpid/cpp/src/qpid/broker/Queue.h
+++ b/qpid/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();
         };
     }
 }