QPID-3243: correctly use --max-queue-count value to compute flow limit.

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

4 files changed, 32 insertions, 4 deletions

diff --git a/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp b/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp
index 5abd7fe666..9f1d3b65f8 100644
--- a/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp
+++ b/qpid/cpp/src/qpid/broker/QueueFlowLimit.cpp
@@ -282,7 +282,9 @@ QueueFlowLimit *QueueFlowLimit::createLimit(Queue *queue, const qpid::framing::F
         return 0;
     }
 
-    if (settings.get(flowStopCountKey) || settings.get(flowStopSizeKey)) {
+    if (settings.get(flowStopCountKey) || settings.get(flowStopSizeKey) ||
+        settings.get(flowResumeCountKey) || settings.get(flowResumeSizeKey)) {
+        // user provided (some) flow settings manually...
         uint32_t flowStopCount = getCapacity(settings, flowStopCountKey, 0);
         uint32_t flowResumeCount = getCapacity(settings, flowResumeCountKey, 0);
         uint64_t flowStopSize = getCapacity(settings, flowStopSizeKey, 0);
@@ -293,11 +295,15 @@ QueueFlowLimit *QueueFlowLimit::createLimit(Queue *queue, const qpid::framing::F
         return new QueueFlowLimit(queue, flowStopCount, flowResumeCount, flowStopSize, flowResumeSize);
     }
 
-    if (defaultFlowStopRatio) {
+    if (defaultFlowStopRatio) {   // broker has a default ratio setup...
         uint64_t maxByteCount = getCapacity(settings, QueuePolicy::maxSizeKey, defaultMaxSize);
         uint64_t flowStopSize = (uint64_t)(maxByteCount * (defaultFlowStopRatio/100.0) + 0.5);
         uint64_t flowResumeSize = (uint64_t)(maxByteCount * (defaultFlowResumeRatio/100.0));
-        return new QueueFlowLimit(queue, 0, 0, flowStopSize, flowResumeSize);
+        uint32_t maxMsgCount = getCapacity(settings, QueuePolicy::maxCountKey, 0);  // no size by default
+        uint32_t flowStopCount = (uint32_t)(maxMsgCount * (defaultFlowStopRatio/100.0) + 0.5);
+        uint32_t flowResumeCount = (uint32_t)(maxMsgCount * (defaultFlowResumeRatio/100.0));
+
+        return new QueueFlowLimit(queue, flowStopCount, flowResumeCount, flowStopSize, flowResumeSize);
     }
     return 0;
 }
diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp
index f9c058c771..5455105078 100644
--- a/qpid/cpp/src/tests/QueuePolicyTest.cpp
+++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp
@@ -386,6 +386,7 @@ QPID_AUTO_TEST_CASE(testCapacityConversion)
 {
     FieldTable args;
     args.setString("qpid.max_count", "5");
+    args.setString("qpid.flow_stop_count", "0");
 
     ProxySessionFixture f;
     std::string q("q");
diff --git a/qpid/cpp/src/tests/cluster_tests.py b/qpid/cpp/src/tests/cluster_tests.py
index 727934ae26..85ace1008a 100755
--- a/qpid/cpp/src/tests/cluster_tests.py
+++ b/qpid/cpp/src/tests/cluster_tests.py
@@ -567,7 +567,7 @@ acl allow all all
         s0 = c0.session()
         # Declare multiple queues bound to same key on amq.topic
         def declare(q,max=0):
-            if max: declare = 'x-declare:{arguments:{"qpid.max_count":%d}}'%max
+            if max: declare = 'x-declare:{arguments:{"qpid.max_count":%d, "qpid.flow_stop_count":0}}'%max
             else: declare = 'x-declare:{}'
             bind='x-bindings:[{queue:%s,key:key,exchange:"amq.topic"}]'%(q)
             s0.sender("%s;{create:always,node:{%s,%s}}" % (q,declare,bind))
diff --git a/qpid/cpp/src/tests/queue_flow_limit_tests.py b/qpid/cpp/src/tests/queue_flow_limit_tests.py
index 6639cdc393..51f91647fb 100644
--- a/qpid/cpp/src/tests/queue_flow_limit_tests.py
+++ b/qpid/cpp/src/tests/queue_flow_limit_tests.py
@@ -129,6 +129,27 @@ class QueueFlowLimitTests(TestBase010):
             self.assertEqual(i.name, "test01")
             self._delete_queue("test01")
 
+            # now verify that the default ratios are applied if max sizing is specified:
+            command = tool + \
+                " --broker-addr=%s:%s " % (self.broker.host, self.broker.port) \
+                + "add queue test02 --max-queue-count=10000 --max-queue-size=1000000"
+            cmd = popen(command)
+            rc = cmd.close()
+            self.assertEqual(rc, None)
+
+            # now verify the settings
+            qs = self.qmf.getObjects(_class="queue")
+            for i in qs:
+                if i.name == "test02":
+                    ## @todo KAG: can't get the flow size from qmf!  Arrgh!
+                    # no way to verify...
+                    #self.assertEqual(i.arguments.get("qpid.flow_resume_count"), 55)
+                    #self.assertEqual(i.arguments.get("qpid.flow_resume_count"), 55)
+                    self.failIf(i.flowStopped)
+                    break;
+            self.assertEqual(i.name, "test02")
+            self._delete_queue("test02")
+
 
     def test_flow_count(self):
         """ Create a queue with count-based flow limit.  Spawn several