From 9ecd69ebc88fb5d82a693e51eef0475c1a6b282e Mon Sep 17 00:00:00 2001 From: Andrew Stitcher Date: Mon, 2 Apr 2007 11:40:48 +0000 Subject: Fix for the most disruptive items in QPID-243. * All #include lines now use '""' rather than '<>' where appropriate. * #include lines within the qpid project use relative includes so that the same path will work in /usr/include when installed as part of the client libraries. * All the source code has now been rearranged to be under src in a directory analogous to the namespace of the classes in it. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@524769 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 89 ++++++++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) create mode 100644 qpid/cpp/src/tests/QueuePolicyTest.cpp (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp new file mode 100644 index 0000000000..5ccc9417cd --- /dev/null +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -0,0 +1,89 @@ + /* + * + * 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 "../broker/QueuePolicy.h" +#include "qpid_test_plugin.h" + +using namespace qpid::broker; +using namespace qpid::framing; + +class QueuePolicyTest : public CppUnit::TestCase +{ + CPPUNIT_TEST_SUITE(QueuePolicyTest); + CPPUNIT_TEST(testCount); + CPPUNIT_TEST(testSize); + CPPUNIT_TEST(testBoth); + CPPUNIT_TEST(testSettings); + CPPUNIT_TEST_SUITE_END(); + + public: + void testCount(){ + QueuePolicy policy(5, 0); + CPPUNIT_ASSERT(!policy.limitExceeded()); + for (int i = 0; i < 5; i++) policy.enqueued(10); + CPPUNIT_ASSERT_EQUAL((uint64_t) 0, policy.getMaxSize()); + CPPUNIT_ASSERT_EQUAL((uint32_t) 5, policy.getMaxCount()); + CPPUNIT_ASSERT(!policy.limitExceeded()); + policy.enqueued(10); + CPPUNIT_ASSERT(policy.limitExceeded()); + policy.dequeued(10); + CPPUNIT_ASSERT(!policy.limitExceeded()); + policy.enqueued(10); + CPPUNIT_ASSERT(policy.limitExceeded()); + } + + void testSize(){ + QueuePolicy policy(0, 50); + for (int i = 0; i < 5; i++) policy.enqueued(10); + CPPUNIT_ASSERT(!policy.limitExceeded()); + policy.enqueued(10); + CPPUNIT_ASSERT(policy.limitExceeded()); + policy.dequeued(10); + CPPUNIT_ASSERT(!policy.limitExceeded()); + policy.enqueued(10); + CPPUNIT_ASSERT(policy.limitExceeded()); + } + + void testBoth(){ + QueuePolicy policy(5, 50); + for (int i = 0; i < 5; i++) policy.enqueued(11); + CPPUNIT_ASSERT(policy.limitExceeded()); + policy.dequeued(20); + CPPUNIT_ASSERT(!policy.limitExceeded());//fails + policy.enqueued(5); + policy.enqueued(10); + CPPUNIT_ASSERT(policy.limitExceeded()); + } + + void testSettings(){ + //test reading and writing the policy from/to field table + FieldTable settings; + QueuePolicy a(101, 303); + a.update(settings); + QueuePolicy b(settings); + CPPUNIT_ASSERT_EQUAL(a.getMaxCount(), b.getMaxCount()); + CPPUNIT_ASSERT_EQUAL(a.getMaxSize(), b.getMaxSize()); + } +}; + +// Make this test suite a plugin. +CPPUNIT_PLUGIN_IMPLEMENT(); +CPPUNIT_TEST_SUITE_REGISTRATION(QueuePolicyTest); + -- cgit v1.2.1 From 26a723475dc6926bde883c8c7f983ee44d8deb01 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Fri, 13 Apr 2007 20:58:27 +0000 Subject: Moved src/ source code to src/qpid directory: - allows rhm package to build consistently against checked-out or installed qpid. - consistent correspondence between source paths and C++ namespaces. - consistent use of #include in source and by users. - allows header files to split over multiple directories, e.g. separating generated code, separating public API from private files. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@528668 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 5ccc9417cd..467f43638f 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -18,7 +18,7 @@ * under the License. * */ -#include "../broker/QueuePolicy.h" +#include "qpid/broker/QueuePolicy.h" #include "qpid_test_plugin.h" using namespace qpid::broker; -- cgit v1.2.1 From 0b1a16b3ef9e0b52aeb1ac2a6c73c150df3b46ee Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Fri, 30 May 2008 08:13:21 +0000 Subject: Convert remaining cppunit tests to boost test framework to reduce dependencies. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@661587 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 110 +++++++++++++++------------------ 1 file changed, 51 insertions(+), 59 deletions(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 467f43638f..db88682010 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -19,71 +19,63 @@ * */ #include "qpid/broker/QueuePolicy.h" -#include "qpid_test_plugin.h" +#include "unit_test.h" using namespace qpid::broker; using namespace qpid::framing; -class QueuePolicyTest : public CppUnit::TestCase -{ - CPPUNIT_TEST_SUITE(QueuePolicyTest); - CPPUNIT_TEST(testCount); - CPPUNIT_TEST(testSize); - CPPUNIT_TEST(testBoth); - CPPUNIT_TEST(testSettings); - CPPUNIT_TEST_SUITE_END(); - - public: - void testCount(){ - QueuePolicy policy(5, 0); - CPPUNIT_ASSERT(!policy.limitExceeded()); - for (int i = 0; i < 5; i++) policy.enqueued(10); - CPPUNIT_ASSERT_EQUAL((uint64_t) 0, policy.getMaxSize()); - CPPUNIT_ASSERT_EQUAL((uint32_t) 5, policy.getMaxCount()); - CPPUNIT_ASSERT(!policy.limitExceeded()); - policy.enqueued(10); - CPPUNIT_ASSERT(policy.limitExceeded()); - policy.dequeued(10); - CPPUNIT_ASSERT(!policy.limitExceeded()); - policy.enqueued(10); - CPPUNIT_ASSERT(policy.limitExceeded()); - } +QPID_AUTO_TEST_SUITE(QueuePolicyTestSuite) - void testSize(){ - QueuePolicy policy(0, 50); - for (int i = 0; i < 5; i++) policy.enqueued(10); - CPPUNIT_ASSERT(!policy.limitExceeded()); - policy.enqueued(10); - CPPUNIT_ASSERT(policy.limitExceeded()); - policy.dequeued(10); - CPPUNIT_ASSERT(!policy.limitExceeded()); - policy.enqueued(10); - CPPUNIT_ASSERT(policy.limitExceeded()); - } +QPID_AUTO_TEST_CASE(testCount) +{ + QueuePolicy policy(5, 0); + BOOST_CHECK(!policy.limitExceeded()); + for (int i = 0; i < 5; i++) policy.enqueued(10); + BOOST_CHECK_EQUAL((uint64_t) 0, policy.getMaxSize()); + BOOST_CHECK_EQUAL((uint32_t) 5, policy.getMaxCount()); + BOOST_CHECK(!policy.limitExceeded()); + policy.enqueued(10); + BOOST_CHECK(policy.limitExceeded()); + policy.dequeued(10); + BOOST_CHECK(!policy.limitExceeded()); + policy.enqueued(10); + BOOST_CHECK(policy.limitExceeded()); +} - void testBoth(){ - QueuePolicy policy(5, 50); - for (int i = 0; i < 5; i++) policy.enqueued(11); - CPPUNIT_ASSERT(policy.limitExceeded()); - policy.dequeued(20); - CPPUNIT_ASSERT(!policy.limitExceeded());//fails - policy.enqueued(5); - policy.enqueued(10); - CPPUNIT_ASSERT(policy.limitExceeded()); - } +QPID_AUTO_TEST_CASE(testSize) +{ + QueuePolicy policy(0, 50); + for (int i = 0; i < 5; i++) policy.enqueued(10); + BOOST_CHECK(!policy.limitExceeded()); + policy.enqueued(10); + BOOST_CHECK(policy.limitExceeded()); + policy.dequeued(10); + BOOST_CHECK(!policy.limitExceeded()); + policy.enqueued(10); + BOOST_CHECK(policy.limitExceeded()); +} - void testSettings(){ - //test reading and writing the policy from/to field table - FieldTable settings; - QueuePolicy a(101, 303); - a.update(settings); - QueuePolicy b(settings); - CPPUNIT_ASSERT_EQUAL(a.getMaxCount(), b.getMaxCount()); - CPPUNIT_ASSERT_EQUAL(a.getMaxSize(), b.getMaxSize()); - } -}; +QPID_AUTO_TEST_CASE(testBoth) +{ + QueuePolicy policy(5, 50); + for (int i = 0; i < 5; i++) policy.enqueued(11); + BOOST_CHECK(policy.limitExceeded()); + policy.dequeued(20); + BOOST_CHECK(!policy.limitExceeded());//fails + policy.enqueued(5); + policy.enqueued(10); + BOOST_CHECK(policy.limitExceeded()); +} -// Make this test suite a plugin. -CPPUNIT_PLUGIN_IMPLEMENT(); -CPPUNIT_TEST_SUITE_REGISTRATION(QueuePolicyTest); +QPID_AUTO_TEST_CASE(testSettings) +{ + //test reading and writing the policy from/to field table + FieldTable settings; + QueuePolicy a(101, 303); + a.update(settings); + QueuePolicy b(settings); + BOOST_CHECK_EQUAL(a.getMaxCount(), b.getMaxCount()); + BOOST_CHECK_EQUAL(a.getMaxSize(), b.getMaxSize()); +} +QPID_AUTO_TEST_SUITE_END() -- cgit v1.2.1 From ecd04f245b9cec7c61b52ac6aa60b9c37fd5749b Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Sun, 21 Sep 2008 20:39:40 +0000 Subject: Refactoring of queue/queue-policy: - moved some logic out of Queue.cpp into QueuePolicy.cpp - moved QueuedMessage definition into its own header file - added checks for requeue and dequeue - split QueuePolicy logic into different sub classes Added ability to request old messages to be discareded to make room for new ones when configured limit has been reached. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@697603 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 147 +++++++++++++++++++++++++-------- 1 file changed, 113 insertions(+), 34 deletions(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index db88682010..4267047c3f 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -19,63 +19,142 @@ * */ #include "qpid/broker/QueuePolicy.h" +#include "qpid/sys/Time.h" #include "unit_test.h" +#include "MessageUtils.h" +#include "BrokerFixture.h" using namespace qpid::broker; +using namespace qpid::client; using namespace qpid::framing; QPID_AUTO_TEST_SUITE(QueuePolicyTestSuite) +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(testCount) { - QueuePolicy policy(5, 0); - BOOST_CHECK(!policy.limitExceeded()); - for (int i = 0; i < 5; i++) policy.enqueued(10); - BOOST_CHECK_EQUAL((uint64_t) 0, policy.getMaxSize()); - BOOST_CHECK_EQUAL((uint32_t) 5, policy.getMaxCount()); - BOOST_CHECK(!policy.limitExceeded()); - policy.enqueued(10); - BOOST_CHECK(policy.limitExceeded()); - policy.dequeued(10); - BOOST_CHECK(!policy.limitExceeded()); - policy.enqueued(10); - BOOST_CHECK(policy.limitExceeded()); + std::auto_ptr policy(QueuePolicy::createQueuePolicy(5, 0)); + BOOST_CHECK_EQUAL((uint64_t) 0, policy->getMaxSize()); + BOOST_CHECK_EQUAL((uint32_t) 5, policy->getMaxCount()); + + QueuedMessage msg = createMessage(10); + for (size_t i = 0; i < 5; i++) { + policy->tryEnqueue(msg); + } + try { + policy->tryEnqueue(msg); + BOOST_FAIL("Policy did not fail on enqueuing sixth message"); + } catch (const ResourceLimitExceededException&) {} + + policy->dequeued(msg); + policy->tryEnqueue(msg); + + try { + policy->tryEnqueue(msg); + BOOST_FAIL("Policy did not fail on enqueuing sixth message (after dequeue)"); + } catch (const ResourceLimitExceededException&) {} } QPID_AUTO_TEST_CASE(testSize) { - QueuePolicy policy(0, 50); - for (int i = 0; i < 5; i++) policy.enqueued(10); - BOOST_CHECK(!policy.limitExceeded()); - policy.enqueued(10); - BOOST_CHECK(policy.limitExceeded()); - policy.dequeued(10); - BOOST_CHECK(!policy.limitExceeded()); - policy.enqueued(10); - BOOST_CHECK(policy.limitExceeded()); + std::auto_ptr policy(QueuePolicy::createQueuePolicy(0, 50)); + QueuedMessage msg = createMessage(10); + + for (size_t i = 0; i < 5; i++) { + policy->tryEnqueue(msg); + } + try { + policy->tryEnqueue(msg); + BOOST_FAIL("Policy did not fail on aggregate size exceeding 50. " << *policy); + } catch (const ResourceLimitExceededException&) {} + + policy->dequeued(msg); + policy->tryEnqueue(msg); + + try { + policy->tryEnqueue(msg); + BOOST_FAIL("Policy did not fail on aggregate size exceeding 50 (after dequeue). " << *policy); + } catch (const ResourceLimitExceededException&) {} } QPID_AUTO_TEST_CASE(testBoth) { - QueuePolicy policy(5, 50); - for (int i = 0; i < 5; i++) policy.enqueued(11); - BOOST_CHECK(policy.limitExceeded()); - policy.dequeued(20); - BOOST_CHECK(!policy.limitExceeded());//fails - policy.enqueued(5); - policy.enqueued(10); - BOOST_CHECK(policy.limitExceeded()); + std::auto_ptr policy(QueuePolicy::createQueuePolicy(5, 50)); + try { + QueuedMessage msg = createMessage(51); + policy->tryEnqueue(msg); + BOOST_FAIL("Policy did not fail on single message exceeding 50. " << *policy); + } catch (const ResourceLimitExceededException&) {} + + std::vector messages; + messages.push_back(createMessage(15)); + messages.push_back(createMessage(10)); + messages.push_back(createMessage(11)); + messages.push_back(createMessage(2)); + messages.push_back(createMessage(7)); + for (size_t i = 0; i < messages.size(); i++) { + policy->tryEnqueue(messages[i]); + } + //size = 45 at this point, count = 5 + try { + QueuedMessage msg = createMessage(5); + policy->tryEnqueue(msg); + BOOST_FAIL("Policy did not fail on count exceeding 6. " << *policy); + } catch (const ResourceLimitExceededException&) {} + try { + QueuedMessage msg = createMessage(10); + policy->tryEnqueue(msg); + BOOST_FAIL("Policy did not fail on aggregate size exceeding 50. " << *policy); + } catch (const ResourceLimitExceededException&) {} + + + policy->dequeued(messages[0]); + try { + QueuedMessage msg = createMessage(20); + policy->tryEnqueue(msg); + } catch (const ResourceLimitExceededException&) { + BOOST_FAIL("Policy failed incorrectly after dequeue. " << *policy); + } } QPID_AUTO_TEST_CASE(testSettings) { //test reading and writing the policy from/to field table + std::auto_ptr a(QueuePolicy::createQueuePolicy(101, 303)); FieldTable settings; - QueuePolicy a(101, 303); - a.update(settings); - QueuePolicy b(settings); - BOOST_CHECK_EQUAL(a.getMaxCount(), b.getMaxCount()); - BOOST_CHECK_EQUAL(a.getMaxSize(), b.getMaxSize()); + a->update(settings); + std::auto_ptr b(QueuePolicy::createQueuePolicy(settings)); + BOOST_CHECK_EQUAL(a->getMaxCount(), b->getMaxCount()); + BOOST_CHECK_EQUAL(a->getMaxSize(), b->getMaxSize()); +} + +QPID_AUTO_TEST_CASE(testRingPolicy) +{ + FieldTable args; + std::auto_ptr policy = QueuePolicy::createQueuePolicy(5, 0, QueuePolicy::RING); + policy->update(args); + + ProxySessionFixture f; + std::string q("my-ring-queue"); + f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args); + for (int i = 0; i < 10; i++) { + f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q)); + } + client::Message msg; + for (int i = 5; i < 10; i++) { + BOOST_CHECK(f.subs.get(msg, q, qpid::sys::TIME_SEC)); + BOOST_CHECK_EQUAL((boost::format("%1%_%2%") % "Message" % (i+1)).str(), msg.getData()); + } + BOOST_CHECK(!f.subs.get(msg, q)); } + QPID_AUTO_TEST_SUITE_END() -- cgit v1.2.1 From 0e9ac83e4bca2c67c8899b6e1a3717e4320c5d4f Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Tue, 23 Sep 2008 01:05:27 +0000 Subject: Additional policy test git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@698054 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 4267047c3f..d7f3fa5628 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -156,5 +156,28 @@ QPID_AUTO_TEST_CASE(testRingPolicy) BOOST_CHECK(!f.subs.get(msg, q)); } +QPID_AUTO_TEST_CASE(testStrictRingPolicy) +{ + FieldTable args; + std::auto_ptr policy = QueuePolicy::createQueuePolicy(5, 0, QueuePolicy::RING_STRICT); + policy->update(args); + + ProxySessionFixture f; + std::string q("my-ring-queue"); + f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args); + LocalQueue incoming(AckPolicy(0));//no automatic acknowledgements + f.subs.subscribe(incoming, q); + for (int i = 0; i < 5; i++) { + f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q)); + } + for (int i = 0; i < 5; i++) { + BOOST_CHECK_EQUAL(incoming.pop().getData(), (boost::format("%1%_%2%") % "Message" % (i+1)).str()); + } + try { + f.session.messageTransfer(arg::content=client::Message("Message_6", q)); + BOOST_FAIL("Transfer should have failed as "); + } catch (const ResourceLimitExceededException&) {} +} + QPID_AUTO_TEST_SUITE_END() -- cgit v1.2.1 From 3751fdb4d6cdf3600aa97b7ec802c37c6c2a6f01 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Thu, 25 Sep 2008 14:16:51 +0000 Subject: Added ScopedSuppressLogging, used to suppress expected error messages in tests. For examples see src/tests/exception_test.cpp git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@698981 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index d7f3fa5628..cf45f554df 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -18,9 +18,11 @@ * under the License. * */ +#include "unit_test.h" +#include "test_tools.h" + #include "qpid/broker/QueuePolicy.h" #include "qpid/sys/Time.h" -#include "unit_test.h" #include "MessageUtils.h" #include "BrokerFixture.h" @@ -174,8 +176,9 @@ QPID_AUTO_TEST_CASE(testStrictRingPolicy) BOOST_CHECK_EQUAL(incoming.pop().getData(), (boost::format("%1%_%2%") % "Message" % (i+1)).str()); } try { + ScopedSuppressLogging sl; // Suppress messages for expected errors. f.session.messageTransfer(arg::content=client::Message("Message_6", q)); - BOOST_FAIL("Transfer should have failed as "); + BOOST_FAIL("expecting ResourceLimitExceededException."); } catch (const ResourceLimitExceededException&) {} } -- cgit v1.2.1 From d2925719fe1db290ff33621d448e5e3a2c2c26eb Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Tue, 21 Oct 2008 23:30:32 +0000 Subject: Refactored DeliveryRecord and delivery path to remove some redundant code. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@706811 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 1 + 1 file changed, 1 insertion(+) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index cf45f554df..f7fe81a709 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -23,6 +23,7 @@ #include "qpid/broker/QueuePolicy.h" #include "qpid/sys/Time.h" +#include "qpid/framing/reply_exceptions.h" #include "MessageUtils.h" #include "BrokerFixture.h" -- cgit v1.2.1 From 10d07002af4b211dfbbc3341a4edb6ec4c2e5cb5 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Sat, 25 Oct 2008 01:55:06 +0000 Subject: Client API change: Centralize access to subscription status, better control of acquire/accept. client/AckPolicy: removed, functionality moved to Subscription and SubscriptionSettings client/SubscriptionSettings: struct aggregates flow control & accept-acquire parameters for subscribe. client/Subscription: represents active subscription. Query settings, unacked messages, manual accept/acquire client/SubscriptionManager: use AcceptMode, AcquireMode enums rather than confusing bools. Issues addressed by the change: - old use of bool for acceptMode was inverted wrt AMQP enum values, bools are confusing. - old AckPolicy was broken - not possible to access the instance associated with an active subscription - old AckPolicy did not provide a way to do manual acquire, only accept. - setting values on SubscriptionManager to apply to subsequent subscriptions is awkward & error-prone, now can use SubscriptionSettings to control on each subscribe individually. - a subscription is a central concept in AMQP, it deserves to be a class. Subscription and SubscriptionSettings provides a single point for future expansion of interactions with a a Subscription. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@707808 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index f7fe81a709..28f555cf6a 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -168,8 +168,10 @@ QPID_AUTO_TEST_CASE(testStrictRingPolicy) ProxySessionFixture f; std::string q("my-ring-queue"); f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args); - LocalQueue incoming(AckPolicy(0));//no automatic acknowledgements - f.subs.subscribe(incoming, q); + LocalQueue incoming; + SubscriptionSettings settings(FlowControl::unlimited()); + settings.autoAck = 0; // no auto ack. + Subscription sub = f.subs.subscribe(incoming, q, settings); for (int i = 0; i < 5; i++) { f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q)); } -- cgit v1.2.1 From 1e1384f5222cd13aee807bd01768fd78aaeab08c Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Mon, 1 Dec 2008 19:49:23 +0000 Subject: QPID-1497: Ensure policy count and size reflect transactionality of dequeues git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@722200 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 57 ++++++++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 28f555cf6a..e72f7b2ce3 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -185,5 +185,62 @@ QPID_AUTO_TEST_CASE(testStrictRingPolicy) } catch (const ResourceLimitExceededException&) {} } +QPID_AUTO_TEST_CASE(testPolicyWithDtx) +{ + FieldTable args; + std::auto_ptr policy = QueuePolicy::createQueuePolicy(5, 0, QueuePolicy::REJECT); + policy->update(args); + + ProxySessionFixture f; + std::string q("my-policy-queue"); + f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args); + LocalQueue incoming; + SubscriptionSettings settings(FlowControl::unlimited()); + settings.autoAck = 0; // no auto ack. + Subscription sub = f.subs.subscribe(incoming, q, settings); + f.session.dtxSelect(); + Xid tx1(1, "test-dtx-mgr", "tx1"); + f.session.dtxStart(arg::xid=tx1); + for (int i = 0; i < 5; i++) { + f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q)); + } + f.session.dtxEnd(arg::xid=tx1); + f.session.dtxCommit(arg::xid=tx1, arg::onePhase=true); + + Xid tx2(1, "test-dtx-mgr", "tx2"); + f.session.dtxStart(arg::xid=tx2); + for (int i = 0; i < 5; i++) { + BOOST_CHECK_EQUAL(incoming.pop().getData(), (boost::format("%1%_%2%") % "Message" % (i+1)).str()); + } + SequenceSet accepting=sub.getUnaccepted(); + f.session.messageAccept(accepting); + f.session.dtxEnd(arg::xid=tx2); + f.session.dtxPrepare(arg::xid=tx2); + f.session.dtxRollback(arg::xid=tx2); + f.session.messageRelease(accepting); + + Xid tx3(1, "test-dtx-mgr", "tx3"); + f.session.dtxStart(arg::xid=tx3); + for (int i = 0; i < 5; i++) { + incoming.pop(); + } + accepting=sub.getUnaccepted(); + f.session.messageAccept(accepting); + f.session.dtxEnd(arg::xid=tx3); + f.session.dtxPrepare(arg::xid=tx3); + + Session other = f.connection.newSession(); + try { + ScopedSuppressLogging sl; // Suppress messages for expected errors. + other.messageTransfer(arg::content=client::Message("Message_6", q)); + BOOST_FAIL("expecting ResourceLimitExceededException."); + } catch (const ResourceLimitExceededException&) {} + + f.session.dtxCommit(arg::xid=tx3); + //now retry and this time should succeed + other = f.connection.newSession(); + other.messageTransfer(arg::content=client::Message("Message_6", q)); +} + QPID_AUTO_TEST_SUITE_END() -- cgit v1.2.1 From 4afd9fb7b3eef066733cdb80b34f55ba2ef1ae6c Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Wed, 7 Jan 2009 20:50:35 +0000 Subject: Ensure that if no store is loaded we don't flow to disk, but revert to rejecting messages. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@732482 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index e72f7b2ce3..6c650169c7 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -22,6 +22,7 @@ #include "test_tools.h" #include "qpid/broker/QueuePolicy.h" +#include "qpid/client/QueueOptions.h" #include "qpid/sys/Time.h" #include "qpid/framing/reply_exceptions.h" #include "MessageUtils.h" @@ -242,5 +243,32 @@ QPID_AUTO_TEST_CASE(testPolicyWithDtx) other.messageTransfer(arg::content=client::Message("Message_6", q)); } +QPID_AUTO_TEST_CASE(testFlowToDiskWithNoStore) +{ + //Ensure that with no store loaded, we don't flow to disk but + //fallback to rejecting messages + QueueOptions args; + args.setSizePolicy(FLOW_TO_DISK, 0, 5); + + ProxySessionFixture f; + std::string q("my-queue"); + f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args); + LocalQueue incoming; + SubscriptionSettings settings(FlowControl::unlimited()); + settings.autoAck = 0; // no auto ack. + Subscription sub = f.subs.subscribe(incoming, q, settings); + for (int i = 0; i < 5; i++) { + f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q)); + } + for (int i = 0; i < 5; i++) { + BOOST_CHECK_EQUAL(incoming.pop().getData(), (boost::format("%1%_%2%") % "Message" % (i+1)).str()); + } + try { + ScopedSuppressLogging sl; // Suppress messages for expected errors. + f.session.messageTransfer(arg::content=client::Message("Message_6", q)); + BOOST_FAIL("expecting ResourceLimitExceededException."); + } catch (const ResourceLimitExceededException&) {} +} + QPID_AUTO_TEST_SUITE_END() -- cgit v1.2.1 From 2e0e8d01aad3d62fcf9a89af16d8532b5ac2540b Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Wed, 25 Feb 2009 11:02:20 +0000 Subject: QPID-1685: Fixed ring queue policy git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@747744 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 9 +++++++++ 1 file changed, 9 insertions(+) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 6c650169c7..7c7f8b7a10 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -158,6 +158,15 @@ QPID_AUTO_TEST_CASE(testRingPolicy) BOOST_CHECK_EQUAL((boost::format("%1%_%2%") % "Message" % (i+1)).str(), msg.getData()); } BOOST_CHECK(!f.subs.get(msg, q)); + + for (int i = 10; i < 20; i++) { + f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q)); + } + for (int i = 15; i < 20; i++) { + BOOST_CHECK(f.subs.get(msg, q, qpid::sys::TIME_SEC)); + BOOST_CHECK_EQUAL((boost::format("%1%_%2%") % "Message" % (i+1)).str(), msg.getData()); + } + BOOST_CHECK(!f.subs.get(msg, q)); } QPID_AUTO_TEST_CASE(testStrictRingPolicy) -- cgit v1.2.1 From ffd20ee19a5fd027e0007c27a12dd402dbeca4f8 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Tue, 14 Jul 2009 14:32:39 +0000 Subject: Add directory to #include git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@793909 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 7c7f8b7a10..33d9fefcfe 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -18,15 +18,15 @@ * under the License. * */ -#include "unit_test.h" -#include "test_tools.h" +#include "tests/unit_test.h" +#include "tests/test_tools.h" #include "qpid/broker/QueuePolicy.h" #include "qpid/client/QueueOptions.h" #include "qpid/sys/Time.h" #include "qpid/framing/reply_exceptions.h" -#include "MessageUtils.h" -#include "BrokerFixture.h" +#include "tests/MessageUtils.h" +#include "tests/BrokerFixture.h" using namespace qpid::broker; using namespace qpid::client; -- cgit v1.2.1 From 795b3bb9e5c033abf33635119694e21e7143fc0a Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Tue, 14 Jul 2009 14:41:22 +0000 Subject: Remove incorrect directory from #include git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@793912 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 33d9fefcfe..7c7f8b7a10 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -18,15 +18,15 @@ * under the License. * */ -#include "tests/unit_test.h" -#include "tests/test_tools.h" +#include "unit_test.h" +#include "test_tools.h" #include "qpid/broker/QueuePolicy.h" #include "qpid/client/QueueOptions.h" #include "qpid/sys/Time.h" #include "qpid/framing/reply_exceptions.h" -#include "tests/MessageUtils.h" -#include "tests/BrokerFixture.h" +#include "MessageUtils.h" +#include "BrokerFixture.h" using namespace qpid::broker; using namespace qpid::client; -- cgit v1.2.1 From 9259c46ecb8c5f3e98441080a26914bdea59bffe Mon Sep 17 00:00:00 2001 From: Andrew Stitcher Date: Wed, 9 Sep 2009 19:46:56 +0000 Subject: Tidied up namespace usage Miscelleneous whitespace fixes git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@813094 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 45 +++++++++++++++++++--------------- 1 file changed, 25 insertions(+), 20 deletions(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 7c7f8b7a10..f40d30b588 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -7,9 +7,9 @@ * 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 @@ -32,6 +32,9 @@ using namespace qpid::broker; using namespace qpid::client; using namespace qpid::framing; +namespace qpid { +namespace tests { + QPID_AUTO_TEST_SUITE(QueuePolicyTestSuite) QueuedMessage createMessage(uint32_t size) @@ -50,11 +53,11 @@ QPID_AUTO_TEST_CASE(testCount) BOOST_CHECK_EQUAL((uint32_t) 5, policy->getMaxCount()); QueuedMessage msg = createMessage(10); - for (size_t i = 0; i < 5; i++) { + for (size_t i = 0; i < 5; i++) { policy->tryEnqueue(msg); } try { - policy->tryEnqueue(msg); + policy->tryEnqueue(msg); BOOST_FAIL("Policy did not fail on enqueuing sixth message"); } catch (const ResourceLimitExceededException&) {} @@ -62,7 +65,7 @@ QPID_AUTO_TEST_CASE(testCount) policy->tryEnqueue(msg); try { - policy->tryEnqueue(msg); + policy->tryEnqueue(msg); BOOST_FAIL("Policy did not fail on enqueuing sixth message (after dequeue)"); } catch (const ResourceLimitExceededException&) {} } @@ -71,12 +74,12 @@ QPID_AUTO_TEST_CASE(testSize) { std::auto_ptr policy(QueuePolicy::createQueuePolicy(0, 50)); QueuedMessage msg = createMessage(10); - - for (size_t i = 0; i < 5; i++) { + + for (size_t i = 0; i < 5; i++) { policy->tryEnqueue(msg); } try { - policy->tryEnqueue(msg); + policy->tryEnqueue(msg); BOOST_FAIL("Policy did not fail on aggregate size exceeding 50. " << *policy); } catch (const ResourceLimitExceededException&) {} @@ -84,7 +87,7 @@ QPID_AUTO_TEST_CASE(testSize) policy->tryEnqueue(msg); try { - policy->tryEnqueue(msg); + policy->tryEnqueue(msg); BOOST_FAIL("Policy did not fail on aggregate size exceeding 50 (after dequeue). " << *policy); } catch (const ResourceLimitExceededException&) {} } @@ -104,7 +107,7 @@ QPID_AUTO_TEST_CASE(testBoth) messages.push_back(createMessage(11)); messages.push_back(createMessage(2)); messages.push_back(createMessage(7)); - for (size_t i = 0; i < messages.size(); i++) { + for (size_t i = 0; i < messages.size(); i++) { policy->tryEnqueue(messages[i]); } //size = 45 at this point, count = 5 @@ -140,7 +143,7 @@ QPID_AUTO_TEST_CASE(testSettings) BOOST_CHECK_EQUAL(a->getMaxSize(), b->getMaxSize()); } -QPID_AUTO_TEST_CASE(testRingPolicy) +QPID_AUTO_TEST_CASE(testRingPolicy) { FieldTable args; std::auto_ptr policy = QueuePolicy::createQueuePolicy(5, 0, QueuePolicy::RING); @@ -169,7 +172,7 @@ QPID_AUTO_TEST_CASE(testRingPolicy) BOOST_CHECK(!f.subs.get(msg, q)); } -QPID_AUTO_TEST_CASE(testStrictRingPolicy) +QPID_AUTO_TEST_CASE(testStrictRingPolicy) { FieldTable args; std::auto_ptr policy = QueuePolicy::createQueuePolicy(5, 0, QueuePolicy::RING_STRICT); @@ -181,7 +184,7 @@ QPID_AUTO_TEST_CASE(testStrictRingPolicy) LocalQueue incoming; SubscriptionSettings settings(FlowControl::unlimited()); settings.autoAck = 0; // no auto ack. - Subscription sub = f.subs.subscribe(incoming, q, settings); + Subscription sub = f.subs.subscribe(incoming, q, settings); for (int i = 0; i < 5; i++) { f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q)); } @@ -192,10 +195,10 @@ QPID_AUTO_TEST_CASE(testStrictRingPolicy) ScopedSuppressLogging sl; // Suppress messages for expected errors. f.session.messageTransfer(arg::content=client::Message("Message_6", q)); BOOST_FAIL("expecting ResourceLimitExceededException."); - } catch (const ResourceLimitExceededException&) {} + } catch (const ResourceLimitExceededException&) {} } -QPID_AUTO_TEST_CASE(testPolicyWithDtx) +QPID_AUTO_TEST_CASE(testPolicyWithDtx) { FieldTable args; std::auto_ptr policy = QueuePolicy::createQueuePolicy(5, 0, QueuePolicy::REJECT); @@ -207,7 +210,7 @@ QPID_AUTO_TEST_CASE(testPolicyWithDtx) LocalQueue incoming; SubscriptionSettings settings(FlowControl::unlimited()); settings.autoAck = 0; // no auto ack. - Subscription sub = f.subs.subscribe(incoming, q, settings); + Subscription sub = f.subs.subscribe(incoming, q, settings); f.session.dtxSelect(); Xid tx1(1, "test-dtx-mgr", "tx1"); f.session.dtxStart(arg::xid=tx1); @@ -244,7 +247,7 @@ QPID_AUTO_TEST_CASE(testPolicyWithDtx) ScopedSuppressLogging sl; // Suppress messages for expected errors. other.messageTransfer(arg::content=client::Message("Message_6", q)); BOOST_FAIL("expecting ResourceLimitExceededException."); - } catch (const ResourceLimitExceededException&) {} + } catch (const ResourceLimitExceededException&) {} f.session.dtxCommit(arg::xid=tx3); //now retry and this time should succeed @@ -252,7 +255,7 @@ QPID_AUTO_TEST_CASE(testPolicyWithDtx) other.messageTransfer(arg::content=client::Message("Message_6", q)); } -QPID_AUTO_TEST_CASE(testFlowToDiskWithNoStore) +QPID_AUTO_TEST_CASE(testFlowToDiskWithNoStore) { //Ensure that with no store loaded, we don't flow to disk but //fallback to rejecting messages @@ -265,7 +268,7 @@ QPID_AUTO_TEST_CASE(testFlowToDiskWithNoStore) LocalQueue incoming; SubscriptionSettings settings(FlowControl::unlimited()); settings.autoAck = 0; // no auto ack. - Subscription sub = f.subs.subscribe(incoming, q, settings); + Subscription sub = f.subs.subscribe(incoming, q, settings); for (int i = 0; i < 5; i++) { f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q)); } @@ -276,8 +279,10 @@ QPID_AUTO_TEST_CASE(testFlowToDiskWithNoStore) ScopedSuppressLogging sl; // Suppress messages for expected errors. f.session.messageTransfer(arg::content=client::Message("Message_6", q)); BOOST_FAIL("expecting ResourceLimitExceededException."); - } catch (const ResourceLimitExceededException&) {} + } catch (const ResourceLimitExceededException&) {} } QPID_AUTO_TEST_SUITE_END() + +}} // namespace qpid::tests -- cgit v1.2.1 From 410f2e510147b502ad3b625572b2a337e550f63b Mon Sep 17 00:00:00 2001 From: Kim van der Riet Date: Tue, 22 Sep 2009 17:36:01 +0000 Subject: Joint checkin from gsim, kpvdr, cctrieloff. See QPID-2102: Exceeding reject queue policy under a transaction causes broker crash git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@817742 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 58 ++++++++++++++++++++++------------ 1 file changed, 37 insertions(+), 21 deletions(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index f40d30b588..875976db85 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -48,56 +48,56 @@ QueuedMessage createMessage(uint32_t size) QPID_AUTO_TEST_CASE(testCount) { - std::auto_ptr policy(QueuePolicy::createQueuePolicy(5, 0)); + std::auto_ptr policy(QueuePolicy::createQueuePolicy("test", 5, 0)); BOOST_CHECK_EQUAL((uint64_t) 0, policy->getMaxSize()); BOOST_CHECK_EQUAL((uint32_t) 5, policy->getMaxCount()); QueuedMessage msg = createMessage(10); for (size_t i = 0; i < 5; i++) { - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); } try { - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); BOOST_FAIL("Policy did not fail on enqueuing sixth message"); } catch (const ResourceLimitExceededException&) {} policy->dequeued(msg); - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); try { - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); BOOST_FAIL("Policy did not fail on enqueuing sixth message (after dequeue)"); } catch (const ResourceLimitExceededException&) {} } QPID_AUTO_TEST_CASE(testSize) { - std::auto_ptr policy(QueuePolicy::createQueuePolicy(0, 50)); + std::auto_ptr policy(QueuePolicy::createQueuePolicy("test", 0, 50)); QueuedMessage msg = createMessage(10); for (size_t i = 0; i < 5; i++) { - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); } try { - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); BOOST_FAIL("Policy did not fail on aggregate size exceeding 50. " << *policy); } catch (const ResourceLimitExceededException&) {} policy->dequeued(msg); - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); try { - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); BOOST_FAIL("Policy did not fail on aggregate size exceeding 50 (after dequeue). " << *policy); } catch (const ResourceLimitExceededException&) {} } QPID_AUTO_TEST_CASE(testBoth) { - std::auto_ptr policy(QueuePolicy::createQueuePolicy(5, 50)); + std::auto_ptr policy(QueuePolicy::createQueuePolicy("test", 5, 50)); try { QueuedMessage msg = createMessage(51); - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); BOOST_FAIL("Policy did not fail on single message exceeding 50. " << *policy); } catch (const ResourceLimitExceededException&) {} @@ -108,17 +108,17 @@ QPID_AUTO_TEST_CASE(testBoth) messages.push_back(createMessage(2)); messages.push_back(createMessage(7)); for (size_t i = 0; i < messages.size(); i++) { - policy->tryEnqueue(messages[i]); + policy->tryEnqueue(messages[i].payload); } //size = 45 at this point, count = 5 try { QueuedMessage msg = createMessage(5); - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); BOOST_FAIL("Policy did not fail on count exceeding 6. " << *policy); } catch (const ResourceLimitExceededException&) {} try { QueuedMessage msg = createMessage(10); - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); BOOST_FAIL("Policy did not fail on aggregate size exceeding 50. " << *policy); } catch (const ResourceLimitExceededException&) {} @@ -126,7 +126,7 @@ QPID_AUTO_TEST_CASE(testBoth) policy->dequeued(messages[0]); try { QueuedMessage msg = createMessage(20); - policy->tryEnqueue(msg); + policy->tryEnqueue(msg.payload); } catch (const ResourceLimitExceededException&) { BOOST_FAIL("Policy failed incorrectly after dequeue. " << *policy); } @@ -135,10 +135,10 @@ QPID_AUTO_TEST_CASE(testBoth) QPID_AUTO_TEST_CASE(testSettings) { //test reading and writing the policy from/to field table - std::auto_ptr a(QueuePolicy::createQueuePolicy(101, 303)); + std::auto_ptr a(QueuePolicy::createQueuePolicy("test", 101, 303)); FieldTable settings; a->update(settings); - std::auto_ptr b(QueuePolicy::createQueuePolicy(settings)); + std::auto_ptr b(QueuePolicy::createQueuePolicy("test", settings)); BOOST_CHECK_EQUAL(a->getMaxCount(), b->getMaxCount()); BOOST_CHECK_EQUAL(a->getMaxSize(), b->getMaxSize()); } @@ -146,7 +146,7 @@ QPID_AUTO_TEST_CASE(testSettings) QPID_AUTO_TEST_CASE(testRingPolicy) { FieldTable args; - std::auto_ptr policy = QueuePolicy::createQueuePolicy(5, 0, QueuePolicy::RING); + std::auto_ptr policy = QueuePolicy::createQueuePolicy("test", 5, 0, QueuePolicy::RING); policy->update(args); ProxySessionFixture f; @@ -175,7 +175,7 @@ QPID_AUTO_TEST_CASE(testRingPolicy) QPID_AUTO_TEST_CASE(testStrictRingPolicy) { FieldTable args; - std::auto_ptr policy = QueuePolicy::createQueuePolicy(5, 0, QueuePolicy::RING_STRICT); + std::auto_ptr policy = QueuePolicy::createQueuePolicy("test", 5, 0, QueuePolicy::RING_STRICT); policy->update(args); ProxySessionFixture f; @@ -201,7 +201,7 @@ QPID_AUTO_TEST_CASE(testStrictRingPolicy) QPID_AUTO_TEST_CASE(testPolicyWithDtx) { FieldTable args; - std::auto_ptr policy = QueuePolicy::createQueuePolicy(5, 0, QueuePolicy::REJECT); + std::auto_ptr policy = QueuePolicy::createQueuePolicy("test", 5, 0, QueuePolicy::REJECT); policy->update(args); ProxySessionFixture f; @@ -282,6 +282,22 @@ QPID_AUTO_TEST_CASE(testFlowToDiskWithNoStore) } catch (const ResourceLimitExceededException&) {} } +QPID_AUTO_TEST_CASE(testPolicyFailureOnCommit) +{ + FieldTable args; + std::auto_ptr policy = QueuePolicy::createQueuePolicy("test", 5, 0, QueuePolicy::REJECT); + policy->update(args); + + ProxySessionFixture f; + std::string q("q"); + f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args); + f.session.txSelect(); + for (int i = 0; i < 10; i++) { + f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q)); + } + ScopedSuppressLogging sl; // Suppress messages for expected errors. + BOOST_CHECK_THROW(f.session.txCommit(), InternalErrorException); +} QPID_AUTO_TEST_SUITE_END() -- cgit v1.2.1 From 98147863bc1e7816eedf6c957d96390ae35ebc60 Mon Sep 17 00:00:00 2001 From: Jonathan Robie Date: Fri, 1 Oct 2010 13:20:54 +0000 Subject: Fixes two bugs for ring queue policies that involve size. - When messages vary in size, now correctly displaces enough smaller messages to make room for the new message. - When a message is larger than maximum queue size, now correctly rejects the message. Resolves JIRA QPID-2338 (https://issues.apache.org/jira/browse/QPID-2338). git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1003531 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 83 +++++++++++++++++++++++++++++++++- 1 file changed, 81 insertions(+), 2 deletions(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 875976db85..5992151e46 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -1,4 +1,4 @@ - /* +/* * * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file @@ -18,6 +18,7 @@ * under the License. * */ +#include #include "unit_test.h" #include "test_tools.h" @@ -143,7 +144,7 @@ QPID_AUTO_TEST_CASE(testSettings) BOOST_CHECK_EQUAL(a->getMaxSize(), b->getMaxSize()); } -QPID_AUTO_TEST_CASE(testRingPolicy) +QPID_AUTO_TEST_CASE(testRingPolicyCount) { FieldTable args; std::auto_ptr policy = QueuePolicy::createQueuePolicy("test", 5, 0, QueuePolicy::RING); @@ -172,6 +173,84 @@ QPID_AUTO_TEST_CASE(testRingPolicy) BOOST_CHECK(!f.subs.get(msg, q)); } +QPID_AUTO_TEST_CASE(testRingPolicySize) +{ + std::string hundredBytes = std::string(100, 'h'); + std::string fourHundredBytes = std::string (400, 'f'); + std::string thousandBytes = std::string(1000, 't'); + + // Ring queue, 500 bytes maxSize + + FieldTable args; + std::auto_ptr policy = QueuePolicy::createQueuePolicy("test", 0, 500, QueuePolicy::RING); + policy->update(args); + + ProxySessionFixture f; + std::string q("my-ring-queue"); + f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args); + + // A. Send messages 0 .. 5, each 100 bytes + + client::Message m(hundredBytes, q); + + for (int i = 0; i < 6; i++) { + std::stringstream id; + id << i; + m.getMessageProperties().setCorrelationId(id.str()); + f.session.messageTransfer(arg::content=m); + } + + // should find 1 .. 5 on the queue, 0 is displaced by 5 + client::Message msg; + for (int i = 1; i < 6; i++) { + std::stringstream id; + id << i; + BOOST_CHECK(f.subs.get(msg, q, qpid::sys::TIME_SEC)); + BOOST_CHECK_EQUAL(msg.getMessageProperties().getCorrelationId(), id.str()); + } + BOOST_CHECK(!f.subs.get(msg, q)); + + // B. Now make sure that one 400 byte message displaces four 100 byte messages + + // Send messages 0 .. 5, each 100 bytes + for (int i = 0; i < 6; i++) { + client::Message m(hundredBytes, q); + std::stringstream id; + id << i; + m.getMessageProperties().setCorrelationId(id.str()); + f.session.messageTransfer(arg::content=m); + } + + // Now send one 400 byte message + client::Message m2(fourHundredBytes, q); + m2.getMessageProperties().setCorrelationId("6"); + f.session.messageTransfer(arg::content=m2); + + // expect to see 5, 6 on the queue + for (int i = 5; i < 7; i++) { + std::stringstream id; + id << i; + BOOST_CHECK(f.subs.get(msg, q, qpid::sys::TIME_SEC)); + BOOST_CHECK_EQUAL(msg.getMessageProperties().getCorrelationId(), id.str()); + } + BOOST_CHECK(!f.subs.get(msg, q)); + + + // C. Try sending a 1000-byte message, should fail - exceeds maxSize of queue + + client::Message m3(thousandBytes, q); + m3.getMessageProperties().setCorrelationId("6"); + try { + ScopedSuppressLogging sl; + f.session.messageTransfer(arg::content=m3); + BOOST_FAIL("Ooops - successfully added a 1000 byte message to a 512 byte ring queue ..."); + } + catch (...) { + } + +} + + QPID_AUTO_TEST_CASE(testStrictRingPolicy) { FieldTable args; -- cgit v1.2.1 From 56571be42e0eaa4d61ffe2e410436dc180d7458b Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Wed, 10 Nov 2010 16:51:16 +0000 Subject: Fix to r1028346: no point in throwing exception after queue has already been created, so just log error and revert to default (do this for floating point values also) git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1033585 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 5992151e46..90af9c7dd9 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/cpp/src/tests/QueuePolicyTest.cpp @@ -378,6 +378,24 @@ QPID_AUTO_TEST_CASE(testPolicyFailureOnCommit) BOOST_CHECK_THROW(f.session.txCommit(), InternalErrorException); } +QPID_AUTO_TEST_CASE(testCapacityConversion) +{ + FieldTable args; + args.setString("qpid.max_count", "5"); + + ProxySessionFixture f; + std::string q("q"); + f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args); + for (int i = 0; i < 5; i++) { + f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q)); + } + try { + ScopedSuppressLogging sl; // Suppress messages for expected errors. + f.session.messageTransfer(arg::content=client::Message("Message_6", q)); + BOOST_FAIL("expecting ResourceLimitExceededException."); + } catch (const ResourceLimitExceededException&) {} +} + QPID_AUTO_TEST_SUITE_END() }} // namespace qpid::tests -- cgit v1.2.1 From 94a18ea7c9f3bc1a88e99f42f21e41a81fd66c79 Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Sat, 19 Feb 2011 15:03:16 +0000 Subject: QPID-2935: merge producer flow control (C++ broker). git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1072356 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') diff --git a/qpid/cpp/src/tests/QueuePolicyTest.cpp b/qpid/cpp/src/tests/QueuePolicyTest.cpp index 90af9c7dd9..f9c058c771 100644 --- a/qpid/cpp/src/tests/QueuePolicyTest.cpp +++ b/qpid/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"); -- cgit v1.2.1 From 1f473a33b3e5b5becd77b14508b06389c2ecd70a Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Tue, 3 May 2011 22:04:51 +0000 Subject: 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 --- qpid/cpp/src/tests/QueuePolicyTest.cpp | 1 + 1 file changed, 1 insertion(+) (limited to 'qpid/cpp/src/tests/QueuePolicyTest.cpp') 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"); -- cgit v1.2.1