QPID-2347: Signal deletion of queue to active subscribers via a resource-deleted exception.

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

1 files changed, 21 insertions, 1 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index 3eb714186c..6e813e936d 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -101,7 +101,8 @@ Queue::Queue(const string& _name, bool _autodelete,
     eventMode(0),
     eventMgr(0),
     insertSeqNo(0),
-    broker(b)
+    broker(b),
+    deleted(false)
 {
     if (parent != 0 && broker != 0)
     {
@@ -291,6 +292,7 @@ void Queue::notifyListener()
 
 bool Queue::getNextMessage(QueuedMessage& m, Consumer::shared_ptr c)
 {
+    checkNotDeleted();
     if (c->preAcquires()) {
         switch (consumeNextMessage(m, c)) {
           case CONSUMED:
@@ -869,6 +871,17 @@ void Queue::destroy()
     }
 }
 
+void Queue::notifyDeleted()
+{
+    QueueListeners::ListenerSet set;
+    {
+        Mutex::ScopedLock locker(messageLock);
+        listeners.snapshot(set);
+        deleted = true;
+    }
+    set.notifyAll();
+}
+
 void Queue::bound(const string& exchange, const string& key,
                   const FieldTable& args)
 {
@@ -1102,3 +1115,10 @@ bool Queue::isEnqueued(const QueuedMessage& msg)
 }
 
 QueueListeners& Queue::getListeners() { return listeners; }
+
+void Queue::checkNotDeleted()
+{
+    if (deleted) {
+        throw ResourceDeletedException(QPID_MSG("Queue " << getName() << " has been deleted."));
+    }
+}