QPID-4348: HA Use independent sequence numbers for identifying messages

Previously HA code used queue sequence numbers to identify messasges.
This assumes that message sequence is identical on primary and backup.

Implementing new features (for example transactions) requires that we tolerate
ordering differences between primary and backups.

This patch introduces a new, queue-scoped HA sequence number managed by the HA
plugin.  The HA ID is set *before* the message is enqueued and assigned a queue
sequence number. This means it is possible to identify messages before they are
enqueued, e.g. messages in an open transaction.

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

1 files changed, 13 insertions, 2 deletions

diff --git a/cpp/src/qpid/broker/Message.cpp b/cpp/src/qpid/broker/Message.cpp
index 1b3f3bd827..c14f534826 100644
--- a/cpp/src/qpid/broker/Message.cpp
+++ b/cpp/src/qpid/broker/Message.cpp
@@ -42,12 +42,18 @@ using std::string;
 namespace qpid {
 namespace broker {
 
-Message::Message() : deliveryCount(-1), publisher(0), expiration(FAR_FUTURE), timestamp(0), isManagementMessage(false) {}
+Message::Message() : deliveryCount(-1), publisher(0), expiration(FAR_FUTURE), timestamp(0),
+                     isManagementMessage(false), replicationId(0)
+{}
+
 Message::Message(boost::intrusive_ptr<Encoding> e, boost::intrusive_ptr<PersistableMessage> p)
-    : encoding(e), persistentContext(p), deliveryCount(-1), publisher(0), expiration(FAR_FUTURE), timestamp(0), isManagementMessage(false)
+    : encoding(e), persistentContext(p), deliveryCount(-1), publisher(0),
+      expiration(FAR_FUTURE), timestamp(0), isManagementMessage(false),
+      replicationId(0)
 {
     if (persistentContext) persistentContext->setIngressCompletion(e);
 }
+
 Message::~Message() {}
 
 
@@ -308,4 +314,9 @@ void Message::processProperties(MapHandler& handler) const
     encoding->processProperties(handler);
 }
 
+uint64_t Message::getReplicationId() const { return replicationId; }
+
+void Message::setReplicationId(framing::SequenceNumber id) { replicationId = id; }
+
+
 }} // namespace qpid::broker