QPID-5736: only block until sent message is settled; signal rejection by throwing exception

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

3 files changed, 15 insertions, 1 deletions

diff --git a/qpid/cpp/include/qpid/messaging/exceptions.h b/qpid/cpp/include/qpid/messaging/exceptions.h
index 6c65f4a889..1ad8014940 100644
--- a/qpid/cpp/include/qpid/messaging/exceptions.h
+++ b/qpid/cpp/include/qpid/messaging/exceptions.h
@@ -137,6 +137,15 @@ struct QPID_MESSAGING_CLASS_EXTERN SendError : public SenderError
 };
 
 /**
+ * Thrown on a synchronous send to indicate that the message being
+ * sent was rejected.
+ */
+struct QPID_MESSAGING_CLASS_EXTERN MessageRejected : public SendError
+{
+    QPID_MESSAGING_EXTERN MessageRejected(const std::string&);
+};
+
+/**
  * Thrown to indicate that the sender attempted to send a message that
  * would result in the target node on the peer exceeding a
  * preconfigured capacity.
diff --git a/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp b/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
index 68a8f85f82..d78269e440 100644
--- a/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
+++ b/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
@@ -399,10 +399,14 @@ void ConnectionContext::send(boost::shared_ptr<SessionContext> ssn, boost::share
     }
     wakeupDriver();
     if (sync && delivery) {
-        while (!delivery->accepted()) {
+        while (!delivery->delivered()) {
             QPID_LOG(debug, "Waiting for confirmation...");
             wait(ssn, snd);//wait until message has been confirmed
         }
+        if (delivery->rejected()) {
+            throw MessageRejected("Message was rejected by peer");
+        }
+
     }
 }
 
diff --git a/qpid/cpp/src/qpid/messaging/exceptions.cpp b/qpid/cpp/src/qpid/messaging/exceptions.cpp
index 5054fdc682..f7bf4aaee0 100644
--- a/qpid/cpp/src/qpid/messaging/exceptions.cpp
+++ b/qpid/cpp/src/qpid/messaging/exceptions.cpp
@@ -44,6 +44,7 @@ NoMessageAvailable::NoMessageAvailable() : FetchError("No message to fetch") {}
 
 SenderError::SenderError(const std::string& msg) : LinkError(msg) {}
 SendError::SendError(const std::string& msg) : SenderError(msg) {}
+MessageRejected::MessageRejected(const std::string& msg) : SendError(msg) {}
 TargetCapacityExceeded::TargetCapacityExceeded(const std::string& msg) : SendError(msg) {}
 
 SessionError::SessionError(const std::string& msg) : MessagingException(msg) {}