QPID-6299: allow ring and lvq behaviours to be combined

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

4 files changed, 13 insertions, 4 deletions

diff --git a/qpid/cpp/src/CMakeLists.txt b/qpid/cpp/src/CMakeLists.txt
index fe7a809cee..3e5165dfb0 100644
--- a/qpid/cpp/src/CMakeLists.txt
+++ b/qpid/cpp/src/CMakeLists.txt
@@ -1077,6 +1077,7 @@ set (qpidbroker_SOURCES
      qpid/broker/IngressCompletion.cpp
      qpid/broker/Link.cpp
      qpid/broker/LinkRegistry.cpp
+     qpid/broker/LossyLvq.cpp
      qpid/broker/LossyQueue.cpp
      qpid/broker/Lvq.cpp
      qpid/broker/Message.cpp
diff --git a/qpid/cpp/src/qpid/broker/LossyQueue.h b/qpid/cpp/src/qpid/broker/LossyQueue.h
index 3e62151d6f..705865f449 100644
--- a/qpid/cpp/src/qpid/broker/LossyQueue.h
+++ b/qpid/cpp/src/qpid/broker/LossyQueue.h
@@ -29,7 +29,7 @@ namespace broker {
 /**
  * Drops messages to prevent a breach of any configured maximum depth.
  */
-class LossyQueue : public Queue
+class LossyQueue : public virtual Queue
 {
   public:
     LossyQueue(const std::string&, const QueueSettings&, MessageStore* const, management::Manageable*, Broker*);
diff --git a/qpid/cpp/src/qpid/broker/Lvq.h b/qpid/cpp/src/qpid/broker/Lvq.h
index 335270a073..26ba2b4914 100644
--- a/qpid/cpp/src/qpid/broker/Lvq.h
+++ b/qpid/cpp/src/qpid/broker/Lvq.h
@@ -32,7 +32,7 @@ class MessageMap;
  * conjunction with the MessageMap class. This requires an existing
  * message to be 'replaced' by a newer message with the same key.
  */
-class Lvq : public Queue
+class Lvq : public virtual Queue
 {
   public:
     Lvq(const std::string&, std::auto_ptr<MessageMap>, const QueueSettings&, MessageStore* const, management::Manageable*, Broker*);
diff --git a/qpid/cpp/src/qpid/broker/QueueFactory.cpp b/qpid/cpp/src/qpid/broker/QueueFactory.cpp
index 8104fff740..16cdea3b0a 100644
--- a/qpid/cpp/src/qpid/broker/QueueFactory.cpp
+++ b/qpid/cpp/src/qpid/broker/QueueFactory.cpp
@@ -22,6 +22,7 @@
 #include "qpid/broker/Broker.h"
 #include "qpid/broker/QueueSettings.h"
 #include "qpid/broker/Queue.h"
+#include "qpid/broker/LossyLvq.h"
 #include "qpid/broker/LossyQueue.h"
 #include "qpid/broker/Lvq.h"
 #include "qpid/broker/Messages.h"
@@ -51,10 +52,17 @@ boost::shared_ptr<Queue> QueueFactory::create(const std::string& name, const Que
     boost::shared_ptr<QueueFlowLimit> flow_ptr(QueueFlowLimit::createLimit(name, settings));
 
     //1. determine Queue type (i.e. whether we are subclassing Queue)
-    // -> if 'ring' policy is in use then subclass
     boost::shared_ptr<Queue> queue;
     if (settings.dropMessagesAtLimit) {
-        queue = boost::shared_ptr<Queue>(new LossyQueue(name, settings, settings.durable ? store : 0, parent, broker));
+        // -> if 'ring' policy is in use then subclass
+        if (settings.lvqKey.size()) {
+            //combination of ring and lvq:
+            std::auto_ptr<MessageMap> map(new MessageMap(settings.lvqKey));
+            queue = boost::shared_ptr<Queue>(new LossyLvq(name, map, settings, settings.durable ? store : 0, parent, broker));
+        } else {
+            //simple ring:
+            queue = boost::shared_ptr<Queue>(new LossyQueue(name, settings, settings.durable ? store : 0, parent, broker));
+        }
     } else if (settings.selfDestructAtLimit) {
         queue = boost::shared_ptr<Queue>(new SelfDestructQueue(name, settings, settings.durable ? store : 0, parent, broker));
     } else if (settings.lvqKey.size()) {