QPID-4591: patch from Ernie Allen to add queue sequence number to messages

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

3 files changed, 12 insertions, 1 deletions

diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp
index 5e806699de..4c9058e78b 100644
--- a/qpid/cpp/src/qpid/broker/Queue.cpp
+++ b/qpid/cpp/src/qpid/broker/Queue.cpp
@@ -755,6 +755,7 @@ void Queue::push(Message& message, bool /*isRecovery*/)
     {
         Mutex::ScopedLock locker(messageLock);
         message.setSequence(++sequence);
+        if (settings.sequencing) message.addAnnotation(settings.sequenceKey, (uint32_t)sequence);
         messages->publish(message);
         listeners.populate(copy);
         observeEnqueue(message, locker);
diff --git a/qpid/cpp/src/qpid/broker/QueueSettings.cpp b/qpid/cpp/src/qpid/broker/QueueSettings.cpp
index 93b832733c..35aa46260c 100644
--- a/qpid/cpp/src/qpid/broker/QueueSettings.cpp
+++ b/qpid/cpp/src/qpid/broker/QueueSettings.cpp
@@ -63,6 +63,7 @@ const std::string LVQ_LEGACY("qpid.last_value_queue");
 const std::string LVQ_LEGACY_KEY("qpid.LVQ_key");
 const std::string LVQ_LEGACY_NOBROWSE("qpid.last_value_queue_no_browse");
 
+const std::string SEQUENCING("qpid.queue_msg_sequence");
 
 bool handleFairshareSetting(const std::string& basename, const std::string& key, const qpid::types::Variant& value, QueueSettings& settings)
 {
@@ -97,7 +98,8 @@ QueueSettings::QueueSettings(bool d, bool a) :
     noLocal(false),
     isBrowseOnly(false),
     autoDeleteDelay(0),
-    alertRepeatInterval(60)
+    alertRepeatInterval(60),
+    sequencing(false)
 {}
 
 bool QueueSettings::handle(const std::string& key, const qpid::types::Variant& value)
@@ -203,6 +205,10 @@ bool QueueSettings::handle(const std::string& key, const qpid::types::Variant& v
     } else if (key == PAGE_FACTOR) {
         pageFactor = value;
         return true;
+    } else if (key == SEQUENCING) {
+        sequenceKey = value.getString();
+        sequencing = !sequenceKey.empty();
+        return true;
     } else if (key == FILTER) {
         filter = value.asString();
         return true;
diff --git a/qpid/cpp/src/qpid/broker/QueueSettings.h b/qpid/cpp/src/qpid/broker/QueueSettings.h
index cf430db76d..19667e93ae 100644
--- a/qpid/cpp/src/qpid/broker/QueueSettings.h
+++ b/qpid/cpp/src/qpid/broker/QueueSettings.h
@@ -83,6 +83,10 @@ struct QueueSettings
     uint64_t maxFileSize;
     uint64_t maxFileCount;
 
+    std::string sequenceKey;
+    // store bool to avoid testing string value
+    bool sequencing;
+
     std::string filter;
 
     //yuck, yuck