QPID-4984: WIP - Merge from trunk r.1534385.

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

1 files changed, 88 insertions, 82 deletions

diff --git a/qpid/cpp/src/tests/QueueFlowLimitTest.cpp b/qpid/cpp/src/tests/QueueFlowLimitTest.cpp
index d305ca452b..7b0a776062 100644
--- a/qpid/cpp/src/tests/QueueFlowLimitTest.cpp
+++ b/qpid/cpp/src/tests/QueueFlowLimitTest.cpp
@@ -77,8 +77,14 @@ public:
 Message createMessage(uint32_t size)
 {
     static uint32_t seqNum;
-    Message msg = MessageUtils::createMessage(qpid::types::Variant::Map(), std::string (size, 'x'));
-    msg.setSequence(++seqNum);
+    //Need to compute what data size is required to make a given
+    //overall size (use one byte of content in test message to ensure
+    //content frame is added)
+    Message test = MessageUtils::createMessage(qpid::types::Variant::Map(), std::string("x"));
+    size_t min = test.getMessageSize() - 1;
+    if (min > size) throw qpid::Exception("Can't create message that small!");
+    Message msg = MessageUtils::createMessage(qpid::types::Variant::Map(), std::string (size - min, 'x'));
+    msg.setSequence(++seqNum);//this doesn't affect message size
     return msg;
 }
 }
@@ -100,18 +106,18 @@ QPID_AUTO_TEST_CASE(testFlowCount)
 
     std::deque<Message> msgs;
     for (size_t i = 0; i < 6; i++) {
-        msgs.push_back(createMessage(10));
+        msgs.push_back(createMessage(100));
         flow->enqueued(msgs.back());
         BOOST_CHECK(!flow->isFlowControlActive());
     }
     BOOST_CHECK(!flow->isFlowControlActive());  // 6 on queue
-    msgs.push_back(createMessage(10));
+    msgs.push_back(createMessage(100));
     flow->enqueued(msgs.back());
     BOOST_CHECK(!flow->isFlowControlActive());  // 7 on queue
-    msgs.push_back(createMessage(10));
+    msgs.push_back(createMessage(100));
     flow->enqueued(msgs.back());
     BOOST_CHECK(flow->isFlowControlActive());   // 8 on queue, ON
-    msgs.push_back(createMessage(10));
+    msgs.push_back(createMessage(100));
     flow->enqueued(msgs.back());
     BOOST_CHECK(flow->isFlowControlActive());   // 9 on queue, no change to flow control
 
@@ -136,69 +142,69 @@ QPID_AUTO_TEST_CASE(testFlowCount)
 QPID_AUTO_TEST_CASE(testFlowSize)
 {
     FieldTable args;
-    args.setUInt64(QueueFlowLimit::flowStopSizeKey, 70);
-    args.setUInt64(QueueFlowLimit::flowResumeSizeKey, 50);
+    args.setUInt64(QueueFlowLimit::flowStopSizeKey, 700);
+    args.setUInt64(QueueFlowLimit::flowResumeSizeKey, 460);
 
     std::auto_ptr<TestFlow> flow(TestFlow::createTestFlow(args));
 
     BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowStopCount());
     BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowResumeCount());
-    BOOST_CHECK_EQUAL((uint32_t) 70, flow->getFlowStopSize());
-    BOOST_CHECK_EQUAL((uint32_t) 50, flow->getFlowResumeSize());
+    BOOST_CHECK_EQUAL((uint32_t) 700, flow->getFlowStopSize());
+    BOOST_CHECK_EQUAL((uint32_t) 460, flow->getFlowResumeSize());
     BOOST_CHECK(!flow->isFlowControlActive());
     BOOST_CHECK(flow->monitorFlowControl());
 
     std::deque<Message> msgs;
     for (size_t i = 0; i < 6; i++) {
-        msgs.push_back(createMessage(10));
+        msgs.push_back(createMessage(100));
         flow->enqueued(msgs.back());
         BOOST_CHECK(!flow->isFlowControlActive());
     }
-    BOOST_CHECK(!flow->isFlowControlActive());  // 60 on queue
+    BOOST_CHECK(!flow->isFlowControlActive());  // 600 on queue
     BOOST_CHECK_EQUAL(6u, flow->getFlowCount());
-    BOOST_CHECK_EQUAL(60u, flow->getFlowSize());
+    BOOST_CHECK_EQUAL(600u, flow->getFlowSize());
 
-    Message msg_9 = createMessage(9);
-    flow->enqueued(msg_9);
-    BOOST_CHECK(!flow->isFlowControlActive());  // 69 on queue
-    Message tinyMsg_1 = createMessage(1);
+    Message msg_50 = createMessage(50);
+    flow->enqueued(msg_50);
+    BOOST_CHECK(!flow->isFlowControlActive());  // 650 on queue
+    Message tinyMsg_1 = createMessage(40);
     flow->enqueued(tinyMsg_1);
-    BOOST_CHECK(!flow->isFlowControlActive());   // 70 on queue
+    BOOST_CHECK(!flow->isFlowControlActive());   // 690 on queue
 
-    Message tinyMsg_2 = createMessage(1);
+    Message tinyMsg_2 = createMessage(40);
     flow->enqueued(tinyMsg_2);
-    BOOST_CHECK(flow->isFlowControlActive());   // 71 on queue, ON
-    msgs.push_back(createMessage(10));
+    BOOST_CHECK(flow->isFlowControlActive());   // 730 on queue, ON
+    msgs.push_back(createMessage(100));
     flow->enqueued(msgs.back());
-    BOOST_CHECK(flow->isFlowControlActive());   // 81 on queue
+    BOOST_CHECK(flow->isFlowControlActive());   // 830 on queue
     BOOST_CHECK_EQUAL(10u, flow->getFlowCount());
-    BOOST_CHECK_EQUAL(81u, flow->getFlowSize());
+    BOOST_CHECK_EQUAL(830u, flow->getFlowSize());
 
     flow->dequeued(msgs.front());
     msgs.pop_front();
-    BOOST_CHECK(flow->isFlowControlActive());   // 71 on queue
+    BOOST_CHECK(flow->isFlowControlActive());   // 730 on queue
     flow->dequeued(msgs.front());
     msgs.pop_front();
-    BOOST_CHECK(flow->isFlowControlActive());   // 61 on queue
+    BOOST_CHECK(flow->isFlowControlActive());   // 630 on queue
     flow->dequeued(msgs.front());
     msgs.pop_front();
-    BOOST_CHECK(flow->isFlowControlActive());   // 51 on queue
+    BOOST_CHECK(flow->isFlowControlActive());   // 530 on queue
 
     flow->dequeued(tinyMsg_1);
-    BOOST_CHECK(flow->isFlowControlActive());   // 50 on queue
+    BOOST_CHECK(flow->isFlowControlActive());   // 490 on queue
     flow->dequeued(tinyMsg_2);
-    BOOST_CHECK(!flow->isFlowControlActive());   // 49 on queue, OFF
+    BOOST_CHECK(!flow->isFlowControlActive());   // 450 on queue, OFF
 
-    flow->dequeued(msg_9);
-    BOOST_CHECK(!flow->isFlowControlActive());  // 40 on queue
+    flow->dequeued(msg_50);
+    BOOST_CHECK(!flow->isFlowControlActive());  // 400 on queue
     flow->dequeued(msgs.front());
     msgs.pop_front();
-    BOOST_CHECK(!flow->isFlowControlActive());  // 30 on queue
+    BOOST_CHECK(!flow->isFlowControlActive());  // 300 on queue
     flow->dequeued(msgs.front());
     msgs.pop_front();
-    BOOST_CHECK(!flow->isFlowControlActive());  // 20 on queue
+    BOOST_CHECK(!flow->isFlowControlActive());  // 200 on queue
     BOOST_CHECK_EQUAL(2u, flow->getFlowCount());
-    BOOST_CHECK_EQUAL(20u, flow->getFlowSize());
+    BOOST_CHECK_EQUAL(200u, flow->getFlowSize());
 }
 
 QPID_AUTO_TEST_CASE(testFlowArgs)
@@ -227,13 +233,13 @@ QPID_AUTO_TEST_CASE(testFlowCombo)
     FieldTable args;
     args.setInt(QueueFlowLimit::flowStopCountKey, 10);
     args.setInt(QueueFlowLimit::flowResumeCountKey, 5);
-    args.setUInt64(QueueFlowLimit::flowStopSizeKey, 200);
-    args.setUInt64(QueueFlowLimit::flowResumeSizeKey, 100);
+    args.setUInt64(QueueFlowLimit::flowStopSizeKey, 2000);
+    args.setUInt64(QueueFlowLimit::flowResumeSizeKey, 1000);
 
-    std::deque<Message> msgs_1;
-    std::deque<Message> msgs_10;
     std::deque<Message> msgs_50;
     std::deque<Message> msgs_100;
+    std::deque<Message> msgs_500;
+    std::deque<Message> msgs_1000;
 
     Message msg;
 
@@ -243,104 +249,104 @@ QPID_AUTO_TEST_CASE(testFlowCombo)
     // verify flow control comes ON when only count passes its stop point.
 
     for (size_t i = 0; i < 10; i++) {
-        msgs_10.push_back(createMessage(10));
-        flow->enqueued(msgs_10.back());
+        msgs_100.push_back(createMessage(100));
+        flow->enqueued(msgs_100.back());
         BOOST_CHECK(!flow->isFlowControlActive());
     }
-    // count:10 size:100
+    // count:10 size:1000
 
-    msgs_1.push_back(createMessage(1));
-    flow->enqueued(msgs_1.back());  // count:11 size: 101  ->ON
+    msgs_50.push_back(createMessage(50));
+    flow->enqueued(msgs_50.back());  // count:11 size: 1050  ->ON
     BOOST_CHECK(flow->isFlowControlActive());
 
     for (size_t i = 0; i < 6; i++) {
-        flow->dequeued(msgs_10.front());
-        msgs_10.pop_front();
+        flow->dequeued(msgs_100.front());
+        msgs_100.pop_front();
         BOOST_CHECK(flow->isFlowControlActive());
     }
-    // count:5 size: 41
+    // count:5 size: 450
 
-    flow->dequeued(msgs_1.front());        // count: 4 size: 40  ->OFF
-    msgs_1.pop_front();
+    flow->dequeued(msgs_50.front());        // count: 4 size: 400  ->OFF
+    msgs_50.pop_front();
     BOOST_CHECK(!flow->isFlowControlActive());
 
     for (size_t i = 0; i < 4; i++) {
-        flow->dequeued(msgs_10.front());
-        msgs_10.pop_front();
+        flow->dequeued(msgs_100.front());
+        msgs_100.pop_front();
         BOOST_CHECK(!flow->isFlowControlActive());
     }
     // count:0 size:0
 
     // verify flow control comes ON when only size passes its stop point.
 
-    msgs_100.push_back(createMessage(100));
-    flow->enqueued(msgs_100.back());  // count:1 size: 100
+    msgs_1000.push_back(createMessage(1000));
+    flow->enqueued(msgs_1000.back());  // count:1 size: 1000
     BOOST_CHECK(!flow->isFlowControlActive());
 
-    msgs_50.push_back(createMessage(50));
-    flow->enqueued(msgs_50.back());   // count:2 size: 150
+    msgs_500.push_back(createMessage(500));
+    flow->enqueued(msgs_500.back());   // count:2 size: 1500
     BOOST_CHECK(!flow->isFlowControlActive());
 
-    msgs_50.push_back(createMessage(50));
-    flow->enqueued(msgs_50.back());   // count:3 size: 200
+    msgs_500.push_back(createMessage(500));
+    flow->enqueued(msgs_500.back());   // count:3 size: 2000
     BOOST_CHECK(!flow->isFlowControlActive());
 
-    msgs_1.push_back(createMessage(1));
-    flow->enqueued(msgs_1.back());   // count:4 size: 201  ->ON
+    msgs_50.push_back(createMessage(50));
+    flow->enqueued(msgs_50.back());   // count:4 size: 2050  ->ON
     BOOST_CHECK(flow->isFlowControlActive());
 
-    flow->dequeued(msgs_100.front());              // count:3 size:101
-    msgs_100.pop_front();
+    flow->dequeued(msgs_1000.front());              // count:3 size:1050
+    msgs_1000.pop_front();
     BOOST_CHECK(flow->isFlowControlActive());
 
-    flow->dequeued(msgs_1.front());                // count:2 size:100
-    msgs_1.pop_front();
+    flow->dequeued(msgs_50.front());                // count:2 size:1000
+    msgs_50.pop_front();
     BOOST_CHECK(flow->isFlowControlActive());
 
-    flow->dequeued(msgs_50.front());               // count:1 size:50  ->OFF
-    msgs_50.pop_front();
+    flow->dequeued(msgs_500.front());               // count:1 size:500  ->OFF
+    msgs_500.pop_front();
     BOOST_CHECK(!flow->isFlowControlActive());
 
     // verify flow control remains ON until both thresholds drop below their
     // resume point.
 
     for (size_t i = 0; i < 8; i++) {
-        msgs_10.push_back(createMessage(10));
-        flow->enqueued(msgs_10.back());
+        msgs_100.push_back(createMessage(100));
+        flow->enqueued(msgs_100.back());
         BOOST_CHECK(!flow->isFlowControlActive());
     }
-    // count:9 size:130
+    // count:9 size:1300
 
-    msgs_10.push_back(createMessage(10));
-    flow->enqueued(msgs_10.back());              // count:10 size: 140
+    msgs_100.push_back(createMessage(100));
+    flow->enqueued(msgs_100.back());              // count:10 size: 1400
     BOOST_CHECK(!flow->isFlowControlActive());
 
-    msgs_1.push_back(createMessage(1));
-    flow->enqueued(msgs_1.back());               // count:11 size: 141  ->ON
+    msgs_50.push_back(createMessage(50));
+    flow->enqueued(msgs_50.back());               // count:11 size: 1450  ->ON
     BOOST_CHECK(flow->isFlowControlActive());
 
-    msgs_100.push_back(createMessage(100));
-    flow->enqueued(msgs_100.back());     // count:12 size: 241  (both thresholds crossed)
+    msgs_1000.push_back(createMessage(1000));
+    flow->enqueued(msgs_1000.back());     // count:12 size: 2450  (both thresholds crossed)
     BOOST_CHECK(flow->isFlowControlActive());
 
-    // at this point: 9@10 + 1@50 + 1@100 + 1@1 == 12@241
+    // at this point: 9@100 + 1@500 + 1@1000 + 1@50 == 12@2450
 
-    flow->dequeued(msgs_50.front());               // count:11 size:191
-    msgs_50.pop_front();
+    flow->dequeued(msgs_500.front());               // count:11 size:1950
+    msgs_500.pop_front();
     BOOST_CHECK(flow->isFlowControlActive());
 
     for (size_t i = 0; i < 9; i++) {
-        flow->dequeued(msgs_10.front());
-        msgs_10.pop_front();
+        flow->dequeued(msgs_100.front());
+        msgs_100.pop_front();
         BOOST_CHECK(flow->isFlowControlActive());
     }
-    // count:2 size:101
-    flow->dequeued(msgs_1.front());                // count:1 size:100
-    msgs_1.pop_front();
+    // count:2 size:1050
+    flow->dequeued(msgs_50.front());                // count:1 size:1000
+    msgs_50.pop_front();
     BOOST_CHECK(flow->isFlowControlActive());   // still active due to size
 
-    flow->dequeued(msgs_100.front());               // count:0 size:0  ->OFF
-    msgs_100.pop_front();
+    flow->dequeued(msgs_1000.front());               // count:0 size:0  ->OFF
+    msgs_1000.pop_front();
     BOOST_CHECK(!flow->isFlowControlActive());
 }