Fix errors when new API Receiver is closed while in use.

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

2 files changed, 24 insertions, 23 deletions

diff --git a/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp b/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp
index b5d7bf78f4..cbf05fc521 100644
--- a/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp
+++ b/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp
@@ -186,22 +186,25 @@ bool IncomingMessages::process(Handler* handler, qpid::sys::Duration duration)
 {
     AbsTime deadline(AbsTime::now(), duration);
     FrameSet::shared_ptr content;
-    for (Duration timeout = duration; incoming->pop(content, timeout); timeout = Duration(AbsTime::now(), deadline)) {
-        if (content->isA<MessageTransferBody>()) {
-            MessageTransfer transfer(content, *this);
-            if (handler && handler->accept(transfer)) {
-                QPID_LOG(debug, "Delivered " << *content->getMethod());
-                return true;
+    try {
+        for (Duration timeout = duration; incoming->pop(content, timeout); timeout = Duration(AbsTime::now(), deadline)) {
+            if (content->isA<MessageTransferBody>()) {
+                MessageTransfer transfer(content, *this);
+                if (handler && handler->accept(transfer)) {
+                    QPID_LOG(debug, "Delivered " << *content->getMethod());
+                    return true;
+                } else {
+                    //received message for another destination, keep for later
+                    QPID_LOG(debug, "Pushed " << *content->getMethod() << " to received queue");
+                    sys::Mutex::ScopedLock l(lock);
+                    received.push_back(content);
+                }
             } else {
-                //received message for another destination, keep for later
-                QPID_LOG(debug, "Pushed " << *content->getMethod() << " to received queue");
-                sys::Mutex::ScopedLock l(lock);
-                received.push_back(content);
+                //TODO: handle other types of commands (e.g. message-accept, message-flow etc)
             }
-        } else {
-            //TODO: handle other types of commands (e.g. message-accept, message-flow etc)
         }
     }
+    catch (const qpid::ClosedException&) {} // Just return false if queue closed.
     return false;
 }
 
diff --git a/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp b/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp
index 9b706ab3de..fb5675c129 100644
--- a/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp
+++ b/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp
@@ -112,21 +112,16 @@ void ReceiverImpl::init(qpid::client::AsyncSession s, AddressResolution& resolve
 {
     sys::Mutex::ScopedLock l(lock);
     session = s;
+    if (state == CANCELLED) return;
     if (state == UNRESOLVED) {
         source = resolver.resolveSource(session, address);
         assert(source.get());
         state = STARTED;
     }
-    if (state == CANCELLED) {
-        source->cancel(session, destination);
-        parent->receiverCancelled(destination);        
-    } else {
-        source->subscribe(session, destination);
-        startFlow(l);
-    }
+    source->subscribe(session, destination);
+    startFlow(l);
 }
 
-
 const std::string& ReceiverImpl::getName() const {
     sys::Mutex::ScopedLock l(lock);
     return destination;
@@ -156,6 +151,10 @@ ReceiverImpl::ReceiverImpl(SessionImpl& p, const std::string& name,
 
 bool ReceiverImpl::getImpl(qpid::messaging::Message& message, qpid::messaging::Duration timeout)
 {
+    {
+        sys::Mutex::ScopedLock l(lock);
+        if (state == CANCELLED) return false;
+    }
     return parent->get(*this, message, timeout);
 }
 
@@ -163,7 +162,7 @@ bool ReceiverImpl::fetchImpl(qpid::messaging::Message& message, qpid::messaging:
 {
     {
         sys::Mutex::ScopedLock l(lock);
-        if (state == CANCELLED) return false;//TODO: or should this be an error?
+        if (state == CANCELLED) return false;
 
         if (capacity == 0 || state != STARTED) {
             session.messageSetFlowMode(destination, FLOW_MODE_CREDIT);
@@ -174,6 +173,7 @@ bool ReceiverImpl::fetchImpl(qpid::messaging::Message& message, qpid::messaging:
     if (getImpl(message, timeout)) {
         return true;
     } else {
+        if (state == CANCELLED) return false; // Might have been closed during get.
         sync(session).messageFlush(destination);
         {
             sys::Mutex::ScopedLock l(lock);
@@ -198,8 +198,6 @@ bool ReceiverImpl::isClosed() const {
     return state == CANCELLED;
 }
 
-
-
 void ReceiverImpl::setCapacityImpl(uint32_t c)
 {
     sys::Mutex::ScopedLock l(lock);