QPID-4935: handle case where peer to sender settles without setting a delivery state

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

2 files changed, 22 insertions, 2 deletions

diff --git a/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp b/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp
index 1d9889c447..0fa97ab933 100644
--- a/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp
+++ b/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp
@@ -95,8 +95,8 @@ SenderContext::Delivery* SenderContext::send(const qpid::messaging::Message& mes
 
 uint32_t SenderContext::processUnsettled()
 {
-    //remove accepted messages from front of deque
-    while (!deliveries.empty() && deliveries.front().accepted()) {
+    //remove messages from front of deque once peer has confirmed receipt
+    while (!deliveries.empty() && deliveries.front().delivered()) {
         deliveries.front().settle();
         deliveries.pop_front();
     }
@@ -453,10 +453,28 @@ void SenderContext::Delivery::send(pn_link_t* sender)
     pn_link_advance(sender);
 }
 
+bool SenderContext::Delivery::delivered()
+{
+    if (pn_delivery_remote_state(token) || pn_delivery_settled(token)) {
+        //TODO: need a better means for signalling outcomes other than accepted
+        if (rejected()) {
+            QPID_LOG(warning, "delivery " << id << " was rejected by peer");
+        } else if (!accepted()) {
+            QPID_LOG(info, "delivery " << id << " was not accepted by peer");
+        }
+        return true;
+    } else {
+        return false;
+    }
+}
 bool SenderContext::Delivery::accepted()
 {
     return pn_delivery_remote_state(token) == PN_ACCEPTED;
 }
+bool SenderContext::Delivery::rejected()
+{
+    return pn_delivery_remote_state(token) == PN_REJECTED;
+}
 void SenderContext::Delivery::settle()
 {
     pn_delivery_settle(token);
diff --git a/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h b/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h
index ba563af2dc..81d306bab3 100644
--- a/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h
+++ b/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h
@@ -53,7 +53,9 @@ class SenderContext
         Delivery(int32_t id);
         void encode(const qpid::messaging::MessageImpl& message, const qpid::messaging::Address&);
         void send(pn_link_t*);
+        bool delivered();
         bool accepted();
+        bool rejected();
         void settle();
       private:
         int32_t id;