QPID-3603: Merge new HA foundations.

Merged from qpid-3603-7. This is basic support for the new HA approach.
For information & limitations see qpid/cpp/design_docs/new-ha-design.txt.

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

1 files changed, 9 insertions, 2 deletions

diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp
index c16ab72876..fc6ada096f 100644
--- a/cpp/src/qpid/cluster/Connection.cpp
+++ b/cpp/src/qpid/cluster/Connection.cpp
@@ -549,7 +549,7 @@ void Connection::deliveryRecord(const string& qname,
         } else {                // Message at original position in original queue
             queue->find(position, m);
         }
-        // FIXME aconway 2011-08-19: removed:
+        // NOTE: removed:
         // if (!m.payload)
         //      throw Exception(QPID_MSG("deliveryRecord no update message"));
         //
@@ -561,7 +561,14 @@ void Connection::deliveryRecord(const string& qname,
         //
     }
 
-    broker::DeliveryRecord dr(m, queue, tag, acquired, accepted, windowing, credit);
+    // If a subscription is cancelled while there are unacked messages for it
+    // there won't be a consumer. Just null it out in this case, it isn't needed.
+    boost::shared_ptr<broker::Consumer> consumer;
+    try { consumer = semanticState().find(tag); }
+    catch(...) {}
+
+    broker::DeliveryRecord dr(
+        m, queue, tag, consumer, acquired, accepted, windowing, credit);
     dr.setId(id);
     if (cancelled) dr.cancel(dr.getTag());
     if (completed) dr.complete();