Set a configurable default size limit on queues

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-10@679462 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 23 insertions, 10 deletions

diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index f008eb23f7..28c7518600 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -40,6 +40,7 @@
 #include "qpid/sys/Poller.h"
 #include "qpid/sys/Dispatcher.h"
 #include "qpid/sys/Thread.h"
+#include "qpid/sys/Time.h"
 #include "qpid/sys/ConnectionInputHandler.h"
 #include "qpid/sys/ConnectionInputHandlerFactory.h"
 #include "qpid/sys/TimeoutHandler.h"
@@ -86,7 +87,8 @@ Broker::Options::Options(const std::string& name) :
     auth(AUTH_DEFAULT),
     realm("QPID"),
     replayFlushLimit(0),
-    replayHardLimit(0)
+    replayHardLimit(0),
+    queueLimit(100*1048576/*100M default limit*/)
 {
     int c = sys::SystemInfo::concurrency();
     workerThreads=c+1;
@@ -109,7 +111,8 @@ Broker::Options::Options(const std::string& name) :
         ("mgmt-enable,m", optValue(enableMgmt,"yes|no"), "Enable Management")
         ("mgmt-pub-interval", optValue(mgmtPubInterval, "SECONDS"), "Management Publish Interval")
         ("auth", optValue(auth, "yes|no"), "Enable authentication, if disabled all incoming connections will be trusted")
-        ("realm", optValue(realm, "REALM"), "Use the given realm when performing authentication");
+        ("realm", optValue(realm, "REALM"), "Use the given realm when performing authentication")
+        ("default-queue-limit", optValue(queueLimit, "BYTES"), "Default maximum size for queues (in bytes)");
 }
 
 const std::string empty;
@@ -166,6 +169,8 @@ Broker::Broker(const Broker::Options& conf) :
         links.setParent     (vhost);
     }
 
+    QueuePolicy::setDefaultMaxSize(conf.queueLimit);
+
     // Early-Initialize plugins
     const Plugin::Plugins& plugins=Plugin::getPlugins();
     for (Plugin::Plugins::const_iterator i = plugins.begin();
diff --git a/cpp/src/qpid/broker/Broker.h b/cpp/src/qpid/broker/Broker.h
index 531817db83..9bbc6aec94 100644
--- a/cpp/src/qpid/broker/Broker.h
+++ b/cpp/src/qpid/broker/Broker.h
@@ -84,6 +84,7 @@ class Broker : public sys::Runnable, public Plugin::Target,
         std::string realm;
         size_t replayFlushLimit;
         size_t replayHardLimit;
+        uint queueLimit;
     };
     
     virtual ~Broker();
diff --git a/cpp/src/qpid/broker/QueuePolicy.cpp b/cpp/src/qpid/broker/QueuePolicy.cpp
index 9b9717def0..de84362f8f 100644
--- a/cpp/src/qpid/broker/QueuePolicy.cpp
+++ b/cpp/src/qpid/broker/QueuePolicy.cpp
@@ -29,7 +29,7 @@ QueuePolicy::QueuePolicy(uint32_t _maxCount, uint64_t _maxSize) :
 
 QueuePolicy::QueuePolicy(const FieldTable& settings) :
     maxCount(getInt(settings, maxCountKey, 0)), 
-    maxSize(getInt(settings, maxSizeKey, 0)), count(0), size(0) {}
+    maxSize(getInt(settings, maxSizeKey, defaultMaxSize)), count(0), size(0) {}
 
 void QueuePolicy::enqueued(uint64_t _size)
 {
@@ -57,14 +57,17 @@ void QueuePolicy::update(FieldTable& settings)
 
 int QueuePolicy::getInt(const FieldTable& settings, const std::string& key, int defaultValue)
 {
-    //Note: currently field table only contain signed 32 bit ints, which
-    //      restricts the values that can be set on the queue policy.
-    try {
-        return settings.getInt(key); 
-    } catch (FieldValueException& ignore) {
-        return defaultValue;
-    }
+    FieldTable::ValuePtr v = settings.get(key);
+    if (v && v->convertsTo<int>()) return v->get<int>();
+    else return defaultValue;
+}
+
+void QueuePolicy::setDefaultMaxSize(uint64_t s)
+{
+    defaultMaxSize = s;
 }
 
 const std::string QueuePolicy::maxCountKey("qpid.max_count");
 const std::string QueuePolicy::maxSizeKey("qpid.max_size");
+uint64_t QueuePolicy::defaultMaxSize(0);
+
diff --git a/cpp/src/qpid/broker/QueuePolicy.h b/cpp/src/qpid/broker/QueuePolicy.h
index 0cc7070e09..2135e327a7 100644
--- a/cpp/src/qpid/broker/QueuePolicy.h
+++ b/cpp/src/qpid/broker/QueuePolicy.h
@@ -29,6 +29,8 @@ namespace qpid {
         {
             static const std::string maxCountKey;
             static const std::string maxSizeKey;
+            
+            static uint64_t defaultMaxSize;
 
             const uint32_t maxCount;
             const uint64_t maxSize;
@@ -46,6 +48,8 @@ namespace qpid {
             bool limitExceeded();
             uint32_t getMaxCount() const { return maxCount; }
             uint64_t getMaxSize() const { return maxSize; }           
+
+            static void setDefaultMaxSize(uint64_t);
         };
     }
 }