QPID-2935: merge producer flow control (C++ broker).

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

12 files changed, 909 insertions, 142 deletions

diff --git a/cpp/src/tests/CMakeLists.txt b/cpp/src/tests/CMakeLists.txt
index 3b3b232671..405718f12b 100644
--- a/cpp/src/tests/CMakeLists.txt
+++ b/cpp/src/tests/CMakeLists.txt
@@ -107,7 +107,6 @@ set(unit_tests_to_build
     MessagingSessionTests
     SequenceSet
     StringUtils
-    IncompleteMessageList
     RangeSet
     AtomicValue
     QueueTest
@@ -119,6 +118,7 @@ set(unit_tests_to_build
     MessageTest
     QueueRegistryTest
     QueuePolicyTest
+    QueueFlowLimitTest
     FramingTest
     HeaderTest
     SequenceNumberTest
diff --git a/cpp/src/tests/IncompleteMessageList.cpp b/cpp/src/tests/IncompleteMessageList.cpp
deleted file mode 100644
index 10782572e5..0000000000
--- a/cpp/src/tests/IncompleteMessageList.cpp
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-#include <iostream>
-#include <sstream>
-#include "qpid/broker/Message.h"
-#include "qpid/broker/NullMessageStore.h"
-#include "qpid/broker/Queue.h"
-#include "qpid/broker/IncompleteMessageList.h"
-
-#include "unit_test.h"
-
-namespace qpid {
-namespace tests {
-
-QPID_AUTO_TEST_SUITE(IncompleteMessageListTestSuite)
-
-using namespace qpid::broker;
-using namespace qpid::framing;
-
-struct Checker
-{
-    std::list<SequenceNumber> ids;
-
-    Checker() { }
-
-    Checker(uint start, uint end) {
-        for (uint i = start; i <= end; i++) {
-            ids.push_back(i);
-        }
-    }
-
-    Checker& expect(const SequenceNumber& id) {
-        ids.push_back(id);
-        return *this;
-    }
-
-    void operator()(boost::intrusive_ptr<Message> msg) {
-        BOOST_CHECK(!ids.empty());
-        BOOST_CHECK_EQUAL(msg->getCommandId(), ids.front());
-        ids.pop_front();
-    }
-};
-
-QPID_AUTO_TEST_CASE(testProcessSimple)
-{
-    IncompleteMessageList list;
-    SequenceNumber counter(1);
-    //fill up list with messages
-    for (int i = 0; i < 5; i++) {
-        boost::intrusive_ptr<Message> msg(new Message(counter++));
-        list.add(msg);
-    }
-    //process and ensure they are all passed to completion listener
-    list.process(Checker(1, 5), false);
-    //process again and ensure none are resent to listener
-    list.process(Checker(), false);
-}
-
-QPID_AUTO_TEST_CASE(testProcessWithIncomplete)
-{
-    Queue::shared_ptr queue;
-    IncompleteMessageList list;
-    SequenceNumber counter(1);
-    boost::intrusive_ptr<Message> middle;
-    //fill up list with messages
-    for (int i = 0; i < 5; i++) {
-        boost::intrusive_ptr<Message> msg(new Message(counter++));
-        list.add(msg);
-        if (i == 2) {
-            //mark a message in the middle as incomplete
-            msg->enqueueAsync(queue, 0);
-            middle = msg;
-        }
-    }
-    //process and ensure only message upto incomplete message are passed to listener
-    list.process(Checker(1, 2), false);
-    //mark message complete and re-process to get remaining messages sent to listener
-    middle->enqueueComplete();
-    list.process(Checker(3, 5), false);
-}
-
-
-struct MockStore : public NullMessageStore
-{
-    Queue::shared_ptr queue;
-    boost::intrusive_ptr<Message> msg;
-
-    void flush(const qpid::broker::PersistableQueue& q) {
-        BOOST_CHECK_EQUAL(queue.get(), &q);
-        msg->enqueueComplete();
-    }
-};
-
-QPID_AUTO_TEST_CASE(testSyncProcessWithIncomplete)
-{
-    IncompleteMessageList list;
-    SequenceNumber counter(1);
-    MockStore store;
-    store.queue = Queue::shared_ptr(new Queue("mock-queue", false, &store));
-    //fill up list with messages
-    for (int i = 0; i < 5; i++) {
-        boost::intrusive_ptr<Message> msg(new Message(counter++));
-        list.add(msg);
-        if (i == 2) {
-            //mark a message in the middle as incomplete
-            msg->enqueueAsync(store.queue, &store);
-            store.msg = msg;
-        }
-    }
-    //process with sync bit specified and ensure that all messages are passed to listener
-    list.process(Checker(1, 5), true);
-}
-
-QPID_AUTO_TEST_SUITE_END()
-
-}} // namespace qpid::tests
diff --git a/cpp/src/tests/Makefile.am b/cpp/src/tests/Makefile.am
index 07405bcd8f..9a1c9e51f6 100644
--- a/cpp/src/tests/Makefile.am
+++ b/cpp/src/tests/Makefile.am
@@ -87,7 +87,6 @@ unit_test_SOURCES= unit_test.cpp unit_test.h \
 	InlineVector.cpp \
 	SequenceSet.cpp \
 	StringUtils.cpp \
-	IncompleteMessageList.cpp \
 	RangeSet.cpp \
 	AtomicValue.cpp \
 	QueueTest.cpp \
@@ -99,6 +98,7 @@ unit_test_SOURCES= unit_test.cpp unit_test.h \
 	MessageTest.cpp \
 	QueueRegistryTest.cpp \
 	QueuePolicyTest.cpp \
+	QueueFlowLimitTest.cpp \
 	FramingTest.cpp \
 	HeaderTest.cpp \
 	SequenceNumberTest.cpp \
@@ -310,7 +310,9 @@ TESTS_ENVIRONMENT = \
     $(srcdir)/run_test 
 
 system_tests = qpid-client-test quick_perftest quick_topictest run_header_test quick_txtest
-TESTS += start_broker $(system_tests) python_tests stop_broker run_federation_tests run_acl_tests run_cli_tests replication_test dynamic_log_level_test
+TESTS += start_broker $(system_tests) python_tests stop_broker run_federation_tests \
+  run_acl_tests run_cli_tests replication_test dynamic_log_level_test \
+  run_queue_flow_limit_tests
 
 EXTRA_DIST +=								\
   run_test vg_check							\
@@ -349,7 +351,8 @@ EXTRA_DIST +=								\
   run_test.ps1								\
   start_broker.ps1							\
   stop_broker.ps1							\
-  topictest.ps1
+  topictest.ps1                                                         \
+  run_queue_flow_limit_tests
 
 check_LTLIBRARIES += libdlclose_noop.la
 libdlclose_noop_la_LDFLAGS = -module -rpath $(abs_builddir)
diff --git a/cpp/src/tests/MessageUtils.h b/cpp/src/tests/MessageUtils.h
index a1b140d484..baca14cf4e 100644
--- a/cpp/src/tests/MessageUtils.h
+++ b/cpp/src/tests/MessageUtils.h
@@ -20,6 +20,7 @@
  */
 
 #include "qpid/broker/Message.h"
+#include "qpid/broker/AsyncCompletion.h"
 #include "qpid/framing/AMQFrame.h"
 #include "qpid/framing/MessageTransferBody.h"
 #include "qpid/framing/Uuid.h"
@@ -28,6 +29,17 @@ using namespace qpid;
 using namespace broker;
 using namespace framing;
 
+namespace {
+    class DummyCompletion : public AsyncCompletion
+    {
+  public:
+        DummyCompletion() {}
+        virtual ~DummyCompletion() {}
+  protected:
+        void completed(bool) {}
+    };
+}
+
 namespace qpid {
 namespace tests {
 
@@ -50,6 +62,8 @@ struct MessageUtils
         msg->getFrames().getHeaders()->get<DeliveryProperties>(true)->setRoutingKey(routingKey);
         if (durable)
             msg->getFrames().getHeaders()->get<DeliveryProperties>(true)->setDeliveryMode(2);
+        boost::shared_ptr<AsyncCompletion>dc(new DummyCompletion());
+        msg->setIngressCompletion(dc);
         return msg;
     }
 
diff --git a/cpp/src/tests/QueueFlowLimitTest.cpp b/cpp/src/tests/QueueFlowLimitTest.cpp
new file mode 100644
index 0000000000..719d6ca6bd
--- /dev/null
+++ b/cpp/src/tests/QueueFlowLimitTest.cpp
@@ -0,0 +1,461 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include <sstream>
+#include <deque>
+#include "unit_test.h"
+#include "test_tools.h"
+
+#include "qpid/broker/QueuePolicy.h"
+#include "qpid/broker/QueueFlowLimit.h"
+#include "qpid/sys/Time.h"
+#include "qpid/framing/reply_exceptions.h"
+#include "MessageUtils.h"
+#include "BrokerFixture.h"
+
+using namespace qpid::broker;
+using namespace qpid::framing;
+
+namespace qpid {
+namespace tests {
+
+QPID_AUTO_TEST_SUITE(QueueFlowLimitTestSuite)
+
+namespace {
+
+class TestFlow : public QueueFlowLimit
+{
+public:
+    TestFlow(uint32_t flowStopCount, uint32_t flowResumeCount,
+             uint64_t flowStopSize, uint64_t flowResumeSize) :
+        QueueFlowLimit(0, flowStopCount, flowResumeCount, flowStopSize, flowResumeSize)
+    {}
+    virtual ~TestFlow() {}
+
+    static TestFlow *createTestFlow(const qpid::framing::FieldTable& settings)
+    {
+        FieldTable::ValuePtr v;
+
+        v = settings.get(flowStopCountKey);
+        uint32_t flowStopCount = (v) ? (uint32_t)v->get<int64_t>() : 0;
+        v = settings.get(flowResumeCountKey);
+        uint32_t flowResumeCount = (v) ? (uint32_t)v->get<int64_t>() : 0;
+        v = settings.get(flowStopSizeKey);
+        uint64_t flowStopSize = (v) ? (uint64_t)v->get<int64_t>() : 0;
+        v = settings.get(flowResumeSizeKey);
+        uint64_t flowResumeSize = (v) ? (uint64_t)v->get<int64_t>() : 0;
+
+        return new TestFlow(flowStopCount, flowResumeCount, flowStopSize, flowResumeSize);
+    }
+
+    static QueueFlowLimit *getQueueFlowLimit(const qpid::framing::FieldTable& settings)
+    {
+        return QueueFlowLimit::createLimit(0, settings);
+    }
+};
+
+
+
+QueuedMessage createMessage(uint32_t size)
+{
+    QueuedMessage msg;
+    msg.payload = MessageUtils::createMessage();
+    MessageUtils::addContent(msg.payload, std::string (size, 'x'));
+    return msg;
+}
+}
+
+QPID_AUTO_TEST_CASE(testFlowCount)
+{
+    FieldTable args;
+    args.setInt(QueueFlowLimit::flowStopCountKey, 7);
+    args.setInt(QueueFlowLimit::flowResumeCountKey, 5);
+
+    std::auto_ptr<TestFlow> flow(TestFlow::createTestFlow(args));
+
+    BOOST_CHECK_EQUAL((uint32_t) 7, flow->getFlowStopCount());
+    BOOST_CHECK_EQUAL((uint32_t) 5, flow->getFlowResumeCount());
+    BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowStopSize());
+    BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowResumeSize());
+    BOOST_CHECK(!flow->isFlowControlActive());
+    BOOST_CHECK(flow->monitorFlowControl());
+
+    std::deque<QueuedMessage> msgs;
+    for (size_t i = 0; i < 6; i++) {
+        msgs.push_back(createMessage(10));
+        flow->enqueued(msgs.back());
+        BOOST_CHECK(!flow->isFlowControlActive());
+    }
+    BOOST_CHECK(!flow->isFlowControlActive());  // 6 on queue
+    msgs.push_back(createMessage(10));
+    flow->enqueued(msgs.back());
+    BOOST_CHECK(!flow->isFlowControlActive());  // 7 on queue
+    msgs.push_back(createMessage(10));
+    flow->enqueued(msgs.back());
+    BOOST_CHECK(flow->isFlowControlActive());   // 8 on queue, ON
+    msgs.push_back(createMessage(10));
+    flow->enqueued(msgs.back());
+    BOOST_CHECK(flow->isFlowControlActive());   // 9 on queue, no change to flow control
+
+    flow->dequeued(msgs.front());
+    msgs.pop_front();
+    BOOST_CHECK(flow->isFlowControlActive());   // 8 on queue
+    flow->dequeued(msgs.front());
+    msgs.pop_front();
+    BOOST_CHECK(flow->isFlowControlActive());   // 7 on queue
+    flow->dequeued(msgs.front());
+    msgs.pop_front();
+    BOOST_CHECK(flow->isFlowControlActive());   // 6 on queue
+    flow->dequeued(msgs.front());
+    msgs.pop_front();
+    BOOST_CHECK(flow->isFlowControlActive());   // 5 on queue, no change
+
+    flow->dequeued(msgs.front());
+    msgs.pop_front();
+    BOOST_CHECK(!flow->isFlowControlActive());  // 4 on queue, OFF
+}
+
+
+QPID_AUTO_TEST_CASE(testFlowSize)
+{
+    FieldTable args;
+    args.setUInt64(QueueFlowLimit::flowStopSizeKey, 70);
+    args.setUInt64(QueueFlowLimit::flowResumeSizeKey, 50);
+
+    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(!flow->isFlowControlActive());
+    BOOST_CHECK(flow->monitorFlowControl());
+
+    std::deque<QueuedMessage> msgs;
+    for (size_t i = 0; i < 6; i++) {
+        msgs.push_back(createMessage(10));
+        flow->enqueued(msgs.back());
+        BOOST_CHECK(!flow->isFlowControlActive());
+    }
+    BOOST_CHECK(!flow->isFlowControlActive());  // 60 on queue
+    BOOST_CHECK_EQUAL(6u, flow->getFlowCount());
+    BOOST_CHECK_EQUAL(60u, flow->getFlowSize());
+
+    QueuedMessage msg_9 = createMessage(9);
+    flow->enqueued(msg_9);
+    BOOST_CHECK(!flow->isFlowControlActive());  // 69 on queue
+    QueuedMessage tinyMsg_1 = createMessage(1);
+    flow->enqueued(tinyMsg_1);
+    BOOST_CHECK(!flow->isFlowControlActive());   // 70 on queue
+
+    QueuedMessage tinyMsg_2 = createMessage(1);
+    flow->enqueued(tinyMsg_2);
+    BOOST_CHECK(flow->isFlowControlActive());   // 71 on queue, ON
+    msgs.push_back(createMessage(10));
+    flow->enqueued(msgs.back());
+    BOOST_CHECK(flow->isFlowControlActive());   // 81 on queue
+    BOOST_CHECK_EQUAL(10u, flow->getFlowCount());
+    BOOST_CHECK_EQUAL(81u, flow->getFlowSize());
+
+    flow->dequeued(msgs.front());
+    msgs.pop_front();
+    BOOST_CHECK(flow->isFlowControlActive());   // 71 on queue
+    flow->dequeued(msgs.front());
+    msgs.pop_front();
+    BOOST_CHECK(flow->isFlowControlActive());   // 61 on queue
+    flow->dequeued(msgs.front());
+    msgs.pop_front();
+    BOOST_CHECK(flow->isFlowControlActive());   // 51 on queue
+
+    flow->dequeued(tinyMsg_1);
+    BOOST_CHECK(flow->isFlowControlActive());   // 50 on queue
+    flow->dequeued(tinyMsg_2);
+    BOOST_CHECK(!flow->isFlowControlActive());   // 49 on queue, OFF
+
+    flow->dequeued(msg_9);
+    BOOST_CHECK(!flow->isFlowControlActive());  // 40 on queue
+    flow->dequeued(msgs.front());
+    msgs.pop_front();
+    BOOST_CHECK(!flow->isFlowControlActive());  // 30 on queue
+    flow->dequeued(msgs.front());
+    msgs.pop_front();
+    BOOST_CHECK(!flow->isFlowControlActive());  // 20 on queue
+    BOOST_CHECK_EQUAL(2u, flow->getFlowCount());
+    BOOST_CHECK_EQUAL(20u, flow->getFlowSize());
+}
+
+QPID_AUTO_TEST_CASE(testFlowArgs)
+{
+    FieldTable args;
+    const uint64_t stop(0x2FFFFFFFF);
+    const uint64_t resume(0x1FFFFFFFF);
+    args.setInt(QueueFlowLimit::flowStopCountKey, 30);
+    args.setInt(QueueFlowLimit::flowResumeCountKey, 21);
+    args.setUInt64(QueueFlowLimit::flowStopSizeKey, stop);
+    args.setUInt64(QueueFlowLimit::flowResumeSizeKey, resume);
+
+    std::auto_ptr<TestFlow> flow(TestFlow::createTestFlow(args));
+
+    BOOST_CHECK_EQUAL((uint32_t) 30, flow->getFlowStopCount());
+    BOOST_CHECK_EQUAL((uint32_t) 21, flow->getFlowResumeCount());
+    BOOST_CHECK_EQUAL(stop, flow->getFlowStopSize());
+    BOOST_CHECK_EQUAL(resume, flow->getFlowResumeSize());
+    BOOST_CHECK(!flow->isFlowControlActive());
+    BOOST_CHECK(flow->monitorFlowControl());
+}
+
+
+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);
+
+    std::deque<QueuedMessage> msgs_1;
+    std::deque<QueuedMessage> msgs_10;
+    std::deque<QueuedMessage> msgs_50;
+    std::deque<QueuedMessage> msgs_100;
+
+    QueuedMessage msg;
+
+    std::auto_ptr<TestFlow> flow(TestFlow::createTestFlow(args));
+    BOOST_CHECK(!flow->isFlowControlActive());        // count:0  size:0
+
+    // 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());
+        BOOST_CHECK(!flow->isFlowControlActive());
+    }
+    // count:10 size:100
+
+    msgs_1.push_back(createMessage(1));
+    flow->enqueued(msgs_1.back());  // count:11 size: 101  ->ON
+    BOOST_CHECK(flow->isFlowControlActive());
+
+    for (size_t i = 0; i < 6; i++) {
+        flow->dequeued(msgs_10.front());
+        msgs_10.pop_front();
+        BOOST_CHECK(flow->isFlowControlActive());
+    }
+    // count:5 size: 41
+
+    flow->dequeued(msgs_1.front());        // count: 4 size: 40  ->OFF
+    msgs_1.pop_front();
+    BOOST_CHECK(!flow->isFlowControlActive());
+
+    for (size_t i = 0; i < 4; i++) {
+        flow->dequeued(msgs_10.front());
+        msgs_10.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
+    BOOST_CHECK(!flow->isFlowControlActive());
+
+    msgs_50.push_back(createMessage(50));
+    flow->enqueued(msgs_50.back());   // count:2 size: 150
+    BOOST_CHECK(!flow->isFlowControlActive());
+
+    msgs_50.push_back(createMessage(50));
+    flow->enqueued(msgs_50.back());   // count:3 size: 200
+    BOOST_CHECK(!flow->isFlowControlActive());
+
+    msgs_1.push_back(createMessage(1));
+    flow->enqueued(msgs_1.back());   // count:4 size: 201  ->ON
+    BOOST_CHECK(flow->isFlowControlActive());
+
+    flow->dequeued(msgs_100.front());              // count:3 size:101
+    msgs_100.pop_front();
+    BOOST_CHECK(flow->isFlowControlActive());
+
+    flow->dequeued(msgs_1.front());                // count:2 size:100
+    msgs_1.pop_front();
+    BOOST_CHECK(flow->isFlowControlActive());
+
+    flow->dequeued(msgs_50.front());               // count:1 size:50  ->OFF
+    msgs_50.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());
+        BOOST_CHECK(!flow->isFlowControlActive());
+    }
+    // count:9 size:130
+
+    msgs_10.push_back(createMessage(10));
+    flow->enqueued(msgs_10.back());              // count:10 size: 140
+    BOOST_CHECK(!flow->isFlowControlActive());
+
+    msgs_1.push_back(createMessage(1));
+    flow->enqueued(msgs_1.back());               // count:11 size: 141  ->ON
+    BOOST_CHECK(flow->isFlowControlActive());
+
+    msgs_100.push_back(createMessage(100));
+    flow->enqueued(msgs_100.back());     // count:12 size: 241  (both thresholds crossed)
+    BOOST_CHECK(flow->isFlowControlActive());
+
+    // at this point: 9@10 + 1@50 + 1@100 + 1@1 == 12@241
+
+    flow->dequeued(msgs_50.front());               // count:11 size:191
+    msgs_50.pop_front();
+    BOOST_CHECK(flow->isFlowControlActive());
+
+    for (size_t i = 0; i < 9; i++) {
+        flow->dequeued(msgs_10.front());
+        msgs_10.pop_front();
+        BOOST_CHECK(flow->isFlowControlActive());
+    }
+    // count:2 size:101
+    flow->dequeued(msgs_1.front());                // count:1 size:100
+    msgs_1.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();
+    BOOST_CHECK(!flow->isFlowControlActive());
+}
+
+
+QPID_AUTO_TEST_CASE(testFlowDefaultArgs)
+{
+    QueueFlowLimit::setDefaults(2950001, // max queue byte count
+                                80,     // 80% stop threshold
+                                70);    // 70% resume threshold
+    FieldTable args;
+    QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args);
+
+    BOOST_CHECK(ptr);
+    std::auto_ptr<QueueFlowLimit> flow(ptr);
+    BOOST_CHECK_EQUAL((uint64_t) 2360001, flow->getFlowStopSize());
+    BOOST_CHECK_EQUAL((uint64_t) 2065000, flow->getFlowResumeSize());
+    BOOST_CHECK_EQUAL( 0u, flow->getFlowStopCount());
+    BOOST_CHECK_EQUAL( 0u, flow->getFlowResumeCount());
+    BOOST_CHECK(!flow->isFlowControlActive());
+    BOOST_CHECK(flow->monitorFlowControl());
+}
+
+
+QPID_AUTO_TEST_CASE(testFlowOverrideArgs)
+{
+    QueueFlowLimit::setDefaults(2950001, // max queue byte count
+                                80,     // 80% stop threshold
+                                70);    // 70% resume threshold
+    {
+        FieldTable args;
+        args.setInt(QueueFlowLimit::flowStopCountKey, 35000);
+        args.setInt(QueueFlowLimit::flowResumeCountKey, 30000);
+
+        QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args);
+        BOOST_CHECK(ptr);
+        std::auto_ptr<QueueFlowLimit> flow(ptr);
+
+        BOOST_CHECK_EQUAL((uint32_t) 35000, flow->getFlowStopCount());
+        BOOST_CHECK_EQUAL((uint32_t) 30000, flow->getFlowResumeCount());
+        BOOST_CHECK_EQUAL((uint64_t) 0, flow->getFlowStopSize());
+        BOOST_CHECK_EQUAL((uint64_t) 0, flow->getFlowResumeSize());
+        BOOST_CHECK(!flow->isFlowControlActive());
+        BOOST_CHECK(flow->monitorFlowControl());
+    }
+    {
+        FieldTable args;
+        args.setInt(QueueFlowLimit::flowStopSizeKey, 350000);
+        args.setInt(QueueFlowLimit::flowResumeSizeKey, 300000);
+
+        QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args);
+        BOOST_CHECK(ptr);
+        std::auto_ptr<QueueFlowLimit> flow(ptr);
+
+        BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowStopCount());
+        BOOST_CHECK_EQUAL((uint32_t) 0, flow->getFlowResumeCount());
+        BOOST_CHECK_EQUAL((uint64_t) 350000, flow->getFlowStopSize());
+        BOOST_CHECK_EQUAL((uint64_t) 300000, flow->getFlowResumeSize());
+        BOOST_CHECK(!flow->isFlowControlActive());
+        BOOST_CHECK(flow->monitorFlowControl());
+    }
+    {
+        FieldTable args;
+        args.setInt(QueueFlowLimit::flowStopCountKey, 35000);
+        args.setInt(QueueFlowLimit::flowResumeCountKey, 30000);
+        args.setInt(QueueFlowLimit::flowStopSizeKey, 350000);
+        args.setInt(QueueFlowLimit::flowResumeSizeKey, 300000);
+
+        QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args);
+        BOOST_CHECK(ptr);
+        std::auto_ptr<QueueFlowLimit> flow(ptr);
+
+        BOOST_CHECK_EQUAL((uint32_t) 35000, flow->getFlowStopCount());
+        BOOST_CHECK_EQUAL((uint32_t) 30000, flow->getFlowResumeCount());
+        BOOST_CHECK_EQUAL((uint64_t) 350000, flow->getFlowStopSize());
+        BOOST_CHECK_EQUAL((uint64_t) 300000, flow->getFlowResumeSize());
+        BOOST_CHECK(!flow->isFlowControlActive());
+        BOOST_CHECK(flow->monitorFlowControl());
+    }
+}
+
+
+QPID_AUTO_TEST_CASE(testFlowOverrideDefaults)
+{
+    QueueFlowLimit::setDefaults(2950001, // max queue byte count
+                                97,     // stop threshold
+                                73);    // resume threshold
+    FieldTable args;
+    QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args);
+    BOOST_CHECK(ptr);
+    std::auto_ptr<QueueFlowLimit> flow(ptr);
+
+    BOOST_CHECK_EQUAL((uint32_t) 2861501, flow->getFlowStopSize());
+    BOOST_CHECK_EQUAL((uint32_t) 2153500, flow->getFlowResumeSize());
+    BOOST_CHECK(!flow->isFlowControlActive());
+    BOOST_CHECK(flow->monitorFlowControl());
+}
+
+
+QPID_AUTO_TEST_CASE(testFlowDisable)
+{
+    {
+        FieldTable args;
+        args.setInt(QueueFlowLimit::flowStopCountKey, 0);
+        QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args);
+        BOOST_CHECK(!ptr);
+    }
+    {
+        FieldTable args;
+        args.setInt(QueueFlowLimit::flowStopSizeKey, 0);
+        QueueFlowLimit *ptr = TestFlow::getQueueFlowLimit(args);
+        BOOST_CHECK(!ptr);
+    }
+}
+
+
+QPID_AUTO_TEST_SUITE_END()
+
+}} // namespace qpid::tests
diff --git a/cpp/src/tests/QueuePolicyTest.cpp b/cpp/src/tests/QueuePolicyTest.cpp
index 90af9c7dd9..f9c058c771 100644
--- a/cpp/src/tests/QueuePolicyTest.cpp
+++ b/cpp/src/tests/QueuePolicyTest.cpp
@@ -23,6 +23,7 @@
 #include "test_tools.h"
 
 #include "qpid/broker/QueuePolicy.h"
+#include "qpid/broker/QueueFlowLimit.h"
 #include "qpid/client/QueueOptions.h"
 #include "qpid/sys/Time.h"
 #include "qpid/framing/reply_exceptions.h"
@@ -38,6 +39,7 @@ namespace tests {
 
 QPID_AUTO_TEST_SUITE(QueuePolicyTestSuite)
 
+namespace {
 QueuedMessage createMessage(uint32_t size)
 {
     QueuedMessage msg;
@@ -45,7 +47,7 @@ QueuedMessage createMessage(uint32_t size)
     MessageUtils::addContent(msg.payload, std::string (size, 'x'));
     return msg;
 }
-
+}
 
 QPID_AUTO_TEST_CASE(testCount)
 {
@@ -340,6 +342,8 @@ QPID_AUTO_TEST_CASE(testFlowToDiskWithNoStore)
     //fallback to rejecting messages
     QueueOptions args;
     args.setSizePolicy(FLOW_TO_DISK, 0, 5);
+    // Disable flow control, or else we'll never hit the max limit
+    args.setInt(QueueFlowLimit::flowStopCountKey, 0);
 
     ProxySessionFixture f;
     std::string q("my-queue");
diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index 57b344498e..e4e9897195 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -36,6 +36,9 @@
 #include "qpid/framing/AMQFrame.h"
 #include "qpid/framing/MessageTransferBody.h"
 #include "qpid/framing/reply_exceptions.h"
+#include "qpid/broker/QueuePolicy.h"
+#include "qpid/broker/QueueFlowLimit.h"
+
 #include <iostream>
 #include "boost/format.hpp"
 
@@ -85,6 +88,8 @@ intrusive_ptr<Message> create_message(std::string exchange, std::string routingK
     msg->getFrames().append(method);
     msg->getFrames().append(header);
     msg->getFrames().getHeaders()->get<DeliveryProperties>(true)->setRoutingKey(routingKey);
+    boost::shared_ptr<AsyncCompletion>dc(new DummyCompletion());
+    msg->setIngressCompletion(dc);
     return msg;
 }
 
@@ -508,6 +513,8 @@ QPID_AUTO_TEST_CASE(testLVQAcquire){
     client::QueueOptions args;
     // set queue mode
     args.setOrdering(client::LVQ);
+    // disable flow control, as this test violates the enqueue/dequeue sequence.
+    args.setInt(QueueFlowLimit::flowStopCountKey, 0);
 
     Queue::shared_ptr queue(new Queue("my-queue", true ));
     queue->configure(args);
diff --git a/cpp/src/tests/TxPublishTest.cpp b/cpp/src/tests/TxPublishTest.cpp
index 6b44d95baa..210abf0a5b 100644
--- a/cpp/src/tests/TxPublishTest.cpp
+++ b/cpp/src/tests/TxPublishTest.cpp
@@ -74,7 +74,7 @@ QPID_AUTO_TEST_CASE(testPrepare)
     BOOST_CHECK_EQUAL(pmsg, t.store.enqueued[0].second);
     BOOST_CHECK_EQUAL(string("queue2"), t.store.enqueued[1].first);
     BOOST_CHECK_EQUAL(pmsg, t.store.enqueued[1].second);
-    BOOST_CHECK_EQUAL( true, ( boost::static_pointer_cast<PersistableMessage>(t.msg))->isEnqueueComplete());
+    BOOST_CHECK_EQUAL( true, ( boost::static_pointer_cast<PersistableMessage>(t.msg))->isIngressComplete());
 }
 
 QPID_AUTO_TEST_CASE(testCommit)
@@ -87,7 +87,7 @@ QPID_AUTO_TEST_CASE(testCommit)
     BOOST_CHECK_EQUAL((uint32_t) 1, t.queue1->getMessageCount());
     intrusive_ptr<Message> msg_dequeue = t.queue1->get().payload;
 
-    BOOST_CHECK_EQUAL( true, (boost::static_pointer_cast<PersistableMessage>(msg_dequeue))->isEnqueueComplete());
+    BOOST_CHECK_EQUAL( true, (boost::static_pointer_cast<PersistableMessage>(msg_dequeue))->isIngressComplete());
     BOOST_CHECK_EQUAL(t.msg, msg_dequeue);
 
     BOOST_CHECK_EQUAL((uint32_t) 1, t.queue2->getMessageCount());
diff --git a/cpp/src/tests/brokertest.py b/cpp/src/tests/brokertest.py
index 98f58ebfdd..6e771bf5d6 100644
--- a/cpp/src/tests/brokertest.py
+++ b/cpp/src/tests/brokertest.py
@@ -29,6 +29,7 @@ from unittest import TestCase
 from copy import copy
 from threading import Thread, Lock, Condition
 from logging import getLogger
+import qmf.console
 
 log = getLogger("qpid.brokertest")
 
@@ -327,6 +328,10 @@ class Broker(Popen):
         log.debug("Started broker %s (%s, %s)" % (self.name, self.pname, self.log))
         self._log_ready = False
 
+    def startQmf(self, handler=None):
+        self.qmf_session = qmf.console.Session(handler)
+        self.qmf_broker = self.qmf_session.addBroker("%s:%s" % (self.host(), self.port()))
+
     def host(self): return self._host
 
     def port(self):
diff --git a/cpp/src/tests/cluster_tests.py b/cpp/src/tests/cluster_tests.py
index cbad4010b4..3e13a3ce8a 100755
--- a/cpp/src/tests/cluster_tests.py
+++ b/cpp/src/tests/cluster_tests.py
@@ -23,7 +23,7 @@ from qpid import datatypes, messaging
 from brokertest import *
 from qpid.harness import Skipped
 from qpid.messaging import Message, Empty
-from threading import Thread, Lock
+from threading import Thread, Lock, Condition
 from logging import getLogger
 from itertools import chain
 from tempfile import NamedTemporaryFile
@@ -304,6 +304,113 @@ acl allow all all
         # Verify logs are consistent
         cluster_test_logs.verify_logs()
 
+    class BlockedSend(Thread):
+        """Send a message, send is expected to block.
+        Verify that it does block (for a given timeout), then allow
+        waiting till it unblocks when it is expected to do so."""
+        def __init__(self, sender, msg):
+            self.sender, self.msg = sender, msg
+            self.blocked = True
+            self.condition = Condition()
+            self.timeout = 0.1    # Time to wait for expected results.
+            Thread.__init__(self)
+        def run(self):
+            try:
+                self.sender.send(self.msg)
+                self.condition.acquire()
+                try:
+                    self.blocked = False
+                    self.condition.notify()
+                finally: self.condition.release()
+            except Exception,e: print "BlockedSend exception: %s"%e
+        def start(self):
+            Thread.start(self)
+            time.sleep(self.timeout)
+            assert self.blocked         # Expected to block
+        def assert_blocked(self): assert self.blocked
+        def wait(self):                 # Now expecting to unblock
+            self.condition.acquire()
+            try:
+                while self.blocked:
+                    self.condition.wait(self.timeout)
+                    if self.blocked: raise Exception("Timed out waiting for send to unblock")
+            finally: self.condition.release()
+            self.join()
+
+    def queue_flowlimit_test(self, brokers):
+        """Verify that the queue's flowlimit configuration and state are
+        correctly replicated.
+        The brokers argument allows this test to run on single broker,
+        cluster of 2 pre-startd brokers or cluster where second broker
+        starts after queue is in flow control.
+        """
+        # configure a queue with a specific flow limit on first broker
+        ssn0 = brokers.first().connect().session()
+        s0 = ssn0.sender("flq; {create:always, node:{type:queue, x-declare:{arguments:{'qpid.flow_stop_count':5, 'qpid.flow_resume_count':3}}}}")
+        brokers.first().startQmf()
+        q = [q for q in brokers.first().qmf_session.getObjects(_class="queue") if q.name == "flq"][0]
+        oid = q.getObjectId()
+        self.assertEqual(q.name, "flq")
+        self.assertEqual(q.arguments, {u'qpid.flow_stop_count': 5L, u'qpid.flow_resume_count': 3L})
+        assert not q.flowStopped
+
+        # fill the queue on one broker until flow control is active
+        for x in range(5): s0.send(Message(str(x)))
+        sender = ShortTests.BlockedSend(s0, Message(str(6)))
+        sender.start()                  # Tests that sender does block
+        # Verify the broker queue goes into a flowStopped state
+        deadline = time.time() + 1
+        while not q.flowStopped and time.time() < deadline: q.update()
+        assert q.flowStopped
+        sender.assert_blocked()         # Still blocked
+
+        # Now verify the  both brokers in cluster have same configuration
+        brokers.second().startQmf()
+        qs = brokers.second().qmf_session.getObjects(_objectId=oid)
+        self.assertEqual(len(qs), 1)
+        q = qs[0]
+        self.assertEqual(q.name, "flq")
+        self.assertEqual(q.arguments, {u'qpid.flow_stop_count': 5L, u'qpid.flow_resume_count': 3L})
+        assert q.flowStopped
+
+        # now drain the queue using a session to the other broker
+        ssn1 = brokers.second().connect().session()
+        r1 = ssn1.receiver("flq", capacity=6)
+        for x in range(4):
+            r1.fetch(timeout=0)
+            ssn1.acknowledge()
+        sender.wait()                   # Verify no longer blocked.
+
+        ssn0.connection.close()
+        ssn1.connection.close()
+        cluster_test_logs.verify_logs()
+
+    def test_queue_flowlimit(self):
+        """Test flow limits on a standalone broker"""
+        broker = self.broker()
+        class Brokers:
+            def first(self): return broker
+            def second(self): return broker
+        self.queue_flowlimit_test(Brokers())
+
+    def test_queue_flowlimit_cluster(self):
+        return          # TODO aconway 2011-02-18: disabled till fixed, QPID-2935
+        cluster = self.cluster(2)
+        class Brokers:
+            def first(self): return cluster[0]
+            def second(self): return cluster[1]
+        self.queue_flowlimit_test(Brokers())
+
+    def test_queue_flowlimit_cluster_join(self):
+        return          # TODO aconway 2011-02-18: disabled till fixed, QPID-2935
+        cluster = self.cluster(1)
+        class Brokers:
+            def first(self): return cluster[0]
+            def second(self):
+                if len(cluster) == 1: cluster.start()
+                return cluster[1]
+        self.queue_flowlimit_test(Brokers())
+
 class LongTests(BrokerTest):
     """Tests that can run for a long time if -DDURATION=<minutes> is set"""
     def duration(self):
diff --git a/cpp/src/tests/queue_flow_limit_tests.py b/cpp/src/tests/queue_flow_limit_tests.py
new file mode 100644
index 0000000000..27fbe2afcc
--- /dev/null
+++ b/cpp/src/tests/queue_flow_limit_tests.py
@@ -0,0 +1,245 @@
+#!/usr/bin/env python
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+# 
+#   http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import sys
+from qpid.testlib import TestBase010
+from qpid import datatypes, messaging
+from qpid.messaging import Message, Empty
+from threading import Thread, Lock
+from logging import getLogger
+from time import sleep
+from subprocess import Popen, PIPE
+from os import environ
+
+class QueueFlowLimitTests(TestBase010):
+
+    def _create_queue(self, name,
+                     stop_count=None, resume_count=None,
+                     stop_size=None, resume_size=None):
+        """ Create a queue with the given flow settings via the queue.declare
+        command.
+        """
+        args={}
+        if (stop_count is not None):
+            args["qpid.flow_stop_count"] = stop_count;
+        if (resume_count is not None):
+            args["qpid.flow_resume_count"] = resume_count;
+        if (stop_size is not None):
+            args["qpid.flow_stop_size"] = stop_size;
+        if (resume_size is not None):
+            args["qpid.flow_resume_size"] = resume_size;
+
+        self.session.queue_declare(queue=name, arguments=args)
+
+        qs = self.qmf.getObjects(_class="queue")
+        for i in qs:
+            if i.name == name:
+                # verify flow settings
+                if (stop_count is not None):
+                    self.assertEqual(i.arguments.get("qpid.flow_stop_count"), stop_count)
+                if (resume_count is not None):
+                    self.assertEqual(i.arguments.get("qpid.flow_resume_count"), resume_count)
+                if (stop_size is not None):
+                    self.assertEqual(i.arguments.get("qpid.flow_stop_size"), stop_size)
+                if (resume_size is not None):
+                    self.assertEqual(i.arguments.get("qpid.flow_resume_size"), resume_size)
+                self.assertFalse(i.flowStopped)
+                return i.getObjectId()
+        self.fail("Unable to create queue '%s'" % name)
+        return None
+
+
+    def _delete_queue(self, name):
+        """ Delete a named queue
+        """
+        self.session.queue_delete(queue=name)
+
+
+    def _start_qpid_send(self, queue, count, content="X", capacity=10):
+        """ Use the qpid-send client to generate traffic to a queue.
+        """
+        command = ["qpid-send",
+                   "-b", "%s:%s" % (self.broker.host, self.broker.port),
+                   "-a", str(queue),
+                   "--messages", str(count),
+                   "--content-string", str(content),
+                   "--capacity", str(capacity)
+                   ]
+
+        return Popen(command, stdout=PIPE)
+
+    def _start_qpid_receive(self, queue, count, timeout=5):
+        """ Use the qpid-receive client to consume from a queue.
+        Note well: prints one line of text to stdout for each consumed msg.
+        """
+        command = ["qpid-receive",
+                   "-b", "%s:%s" % (self.broker.host, self.broker.port),
+                   "-a", str(queue),
+                   "--messages", str(count),
+                   "--timeout", str(timeout),
+                   "--print-content", "yes"
+                   ]
+        return Popen(command, stdout=PIPE)
+
+
+
+    def test_qpid_config_cmd(self):
+        """ Test the qpid-config command's ability to configure a queue's flow
+        control thresholds.
+        """
+        tool = environ.get("QPID_CONFIG_EXEC")
+        if tool:
+            command = [tool,
+                       "--broker-addr=%s:%s" % (self.broker.host, self.broker.port),
+                       "add", "queue", "test01",
+                       "--flow-stop-count=999",
+                       "--flow-resume-count=55",
+                       "--flow-stop-size=5000000",
+                       "--flow-resume-size=100000"]
+            #cmd = Popen(command, stdout=PIPE)
+            cmd = Popen(command)
+            cmd.wait()
+            self.assertEqual(cmd.returncode, 0)
+
+            # now verify the settings
+            self.startQmf();
+            qs = self.qmf.getObjects(_class="queue")
+            for i in qs:
+                if i.name == "test01":
+                    self.assertEqual(i.arguments.get("qpid.flow_stop_count"), 999)
+                    self.assertEqual(i.arguments.get("qpid.flow_resume_count"), 55)
+                    self.assertEqual(i.arguments.get("qpid.flow_stop_size"), 5000000)
+                    self.assertEqual(i.arguments.get("qpid.flow_resume_size"), 100000)
+                    self.assertFalse(i.flowStopped)
+                    break;
+            self.assertEqual(i.name, "test01")
+            self._delete_queue("test01")
+
+
+    def test_flow_count(self):
+        """ Create a queue with count-based flow limit.  Spawn several
+        producers which will exceed the limit.  Verify limit exceeded.  Consume
+        all messages.  Verify flow control released.
+        """
+        self.startQmf();
+        oid = self._create_queue("test-q", stop_count=373, resume_count=229)
+
+        sndr1 = self._start_qpid_send("test-q", count=1213, content="XXX", capacity=50);
+        sndr2 = self._start_qpid_send("test-q", count=797, content="Y", capacity=13);
+        sndr3 = self._start_qpid_send("test-q", count=331, content="ZZZZZ", capacity=149);
+        totalMsgs = 1213 + 797 + 331
+        
+
+        # wait until flow control is active
+        count = 0
+        while self.qmf.getObjects(_objectId=oid)[0].flowStopped == False and \
+                count < 10:
+            sleep(1);
+            count += 1;
+        self.assertTrue(self.qmf.getObjects(_objectId=oid)[0].flowStopped)
+        depth = self.qmf.getObjects(_objectId=oid)[0].msgDepth
+        self.assertGreater(depth, 373)
+
+        # now wait until the enqueues stop happening - ensure that
+        # not all msgs have been sent (senders are blocked)
+        sleep(1)
+        newDepth = self.qmf.getObjects(_objectId=oid)[0].msgDepth
+        while depth != newDepth:
+            depth = newDepth;
+            sleep(1)
+            newDepth = self.qmf.getObjects(_objectId=oid)[0].msgDepth
+        self.assertGreater(totalMsgs, depth)
+
+        # drain the queue
+        rcvr = self._start_qpid_receive("test-q",
+                                        count=totalMsgs)
+        count = 0;
+        x = rcvr.stdout.readline()    # prints a line for each received msg
+        while x:
+            count += 1;
+            x = rcvr.stdout.readline()
+
+        sndr1.wait();
+        sndr2.wait();
+        sndr3.wait();
+        rcvr.wait();
+
+        self.assertEqual(count, totalMsgs)
+        self.assertFalse(self.qmf.getObjects(_objectId=oid)[0].flowStopped)
+
+        self._delete_queue("test-q")
+
+
+    def test_flow_size(self):
+        """ Create a queue with size-based flow limit.  Spawn several
+        producers which will exceed the limit.  Verify limit exceeded.  Consume
+        all messages.  Verify flow control released.
+        """
+        self.startQmf();
+        oid = self._create_queue("test-q", stop_size=351133, resume_size=251143)
+
+        sndr1 = self._start_qpid_send("test-q", count=1699, content="X"*439, capacity=53);
+        sndr2 = self._start_qpid_send("test-q", count=1129, content="Y"*631, capacity=13);
+        sndr3 = self._start_qpid_send("test-q", count=881, content="Z"*823, capacity=149);
+        totalMsgs = 1699 + 1129 + 881
+        totalBytes = 439 + 631 + 823
+
+        # wait until flow control is active
+        count = 0
+        while self.qmf.getObjects(_objectId=oid)[0].flowStopped == False and \
+                count < 10:
+            sleep(1);
+            count += 1;
+        self.assertTrue(self.qmf.getObjects(_objectId=oid)[0].flowStopped)
+        self.assertGreater(self.qmf.getObjects(_objectId=oid)[0].byteDepth, 351133)
+
+        # now wait until the enqueues stop happening - ensure that
+        # not all msgs have been sent (senders are blocked)
+        depth = self.qmf.getObjects(_objectId=oid)[0].msgDepth
+        sleep(1)
+        newDepth = self.qmf.getObjects(_objectId=oid)[0].msgDepth
+        while depth != newDepth:
+            depth = newDepth;
+            sleep(1)
+            newDepth = self.qmf.getObjects(_objectId=oid)[0].msgDepth
+        self.assertGreater(totalMsgs, depth)
+
+        # drain the queue
+        rcvr = self._start_qpid_receive("test-q",
+                                        count=totalMsgs)
+        count = 0;
+        x = rcvr.stdout.readline()    # prints a line for each received msg
+        while x:
+            count += 1;
+            x = rcvr.stdout.readline()
+
+        sndr1.wait();
+        sndr2.wait();
+        sndr3.wait();
+        rcvr.wait();
+
+        self.assertEqual(count, totalMsgs)
+        self.assertFalse(self.qmf.getObjects(_objectId=oid)[0].flowStopped)
+
+        self._delete_queue("test-q")
+
+
+
+
diff --git a/cpp/src/tests/run_queue_flow_limit_tests b/cpp/src/tests/run_queue_flow_limit_tests
new file mode 100755
index 0000000000..9f2f093353
--- /dev/null
+++ b/cpp/src/tests/run_queue_flow_limit_tests
@@ -0,0 +1,55 @@
+#!/bin/sh
+
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+# 
+#   http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# Run tests against Queue producer flow control.
+
+source ./test_env.sh
+test -d $PYTHON_DIR || { echo "Skipping queue flow control tests, no python dir."; exit 0; }
+
+LOG_FILE=queue_flow_limit_test.log
+PORT=""
+
+trap stop_broker INT TERM QUIT
+
+error() {
+    echo $*
+    exit 1;
+}
+
+start_broker() {
+    rm -rf $LOG_FILE
+    PORT=$($QPIDD_EXEC --auth=no --no-module-dir --daemon --port=0 -t --log-to-file $LOG_FILE) || error "Could not start broker"
+}
+
+stop_broker() {
+    test -n "$PORT" && $QPIDD_EXEC --no-module-dir --quit --port $PORT
+}
+
+start_broker
+echo "Running Queue flow limit tests using broker on port $PORT"
+$QPID_PYTHON_TEST -m queue_flow_limit_tests $SKIPTESTS -b localhost:$PORT $@
+RETCODE=$?
+stop_broker
+if test x$RETCODE != x0; then
+    echo "FAIL queue flow limit tests"; exit 1;
+fi
+rm -rf $LOG_FILE
+