QPID-4966: ensure timeout is honoured while waiting of rmessages

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

2 files changed, 33 insertions, 3 deletions

diff --git a/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp b/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
index 4553ebddb3..72bba608d1 100644
--- a/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
+++ b/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
@@ -270,7 +270,7 @@ bool ConnectionContext::get(boost::shared_ptr<SessionContext> ssn, boost::shared
             pn_link_advance(lnk->receiver);
             return true;
         } else if (until > qpid::sys::now()) {
-            wait(ssn, lnk);
+            waitUntil(ssn, lnk, until);
         } else {
             return false;
         }
@@ -427,9 +427,8 @@ pn_state_t REQUIRES_CLOSE = PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED;
 pn_state_t IS_CLOSED = PN_LOCAL_CLOSED | PN_REMOTE_CLOSED;
 }
 
-void ConnectionContext::wait()
+void ConnectionContext::check()
 {
-    lock.wait();
     if (state == DISCONNECTED) {
         throw qpid::messaging::TransportFailure("Disconnected");
     }
@@ -438,6 +437,17 @@ void ConnectionContext::wait()
         throw qpid::messaging::ConnectionError("Connection closed by peer");
     }
 }
+
+void ConnectionContext::wait()
+{
+    lock.wait();
+    check();
+}
+void ConnectionContext::waitUntil(qpid::sys::AbsTime until)
+{
+    lock.wait(until);
+    check();
+}
 void ConnectionContext::wait(boost::shared_ptr<SessionContext> ssn)
 {
     wait();
@@ -453,6 +463,21 @@ void ConnectionContext::wait(boost::shared_ptr<SessionContext> ssn, boost::share
     wait();
     checkClosed(ssn, lnk);
 }
+void ConnectionContext::waitUntil(boost::shared_ptr<SessionContext> ssn, qpid::sys::AbsTime until)
+{
+    waitUntil(until);
+    checkClosed(ssn);
+}
+void ConnectionContext::waitUntil(boost::shared_ptr<SessionContext> ssn, boost::shared_ptr<ReceiverContext> lnk, qpid::sys::AbsTime until)
+{
+    waitUntil(until);
+    checkClosed(ssn, lnk);
+}
+void ConnectionContext::waitUntil(boost::shared_ptr<SessionContext> ssn, boost::shared_ptr<SenderContext> lnk, qpid::sys::AbsTime until)
+{
+    waitUntil(until);
+    checkClosed(ssn, lnk);
+}
 void ConnectionContext::checkClosed(boost::shared_ptr<SessionContext> ssn)
 {
     if ((pn_session_state(ssn->session) & REQUIRES_CLOSE) == REQUIRES_CLOSE) {
diff --git a/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.h b/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.h
index 5627bd903d..37d73ea456 100644
--- a/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.h
+++ b/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.h
@@ -135,10 +135,15 @@ class ConnectionContext : public qpid::sys::ConnectionCodec, public qpid::messag
     };
     CodecSwitch codecSwitch;
 
+    void check();
     void wait();
+    void waitUntil(qpid::sys::AbsTime until);
     void wait(boost::shared_ptr<SessionContext>);
+    void waitUntil(boost::shared_ptr<SessionContext>, qpid::sys::AbsTime until);
     void wait(boost::shared_ptr<SessionContext>, boost::shared_ptr<ReceiverContext>);
     void wait(boost::shared_ptr<SessionContext>, boost::shared_ptr<SenderContext>);
+    void waitUntil(boost::shared_ptr<SessionContext>, boost::shared_ptr<ReceiverContext>, qpid::sys::AbsTime until);
+    void waitUntil(boost::shared_ptr<SessionContext>, boost::shared_ptr<SenderContext>, qpid::sys::AbsTime until);
     void checkClosed(boost::shared_ptr<SessionContext>);
     void checkClosed(boost::shared_ptr<SessionContext>, boost::shared_ptr<ReceiverContext>);
     void checkClosed(boost::shared_ptr<SessionContext>, boost::shared_ptr<SenderContext>);