From c619794e8a903e716bc5117179ea0ab1e24e1254 Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Wed, 17 Oct 2007 08:59:44 +0000 Subject: Use shared pointers for consumers (held by queues and sessions) to prevent having to hold lock across deliver() while avoiding invocation on stale pointers. Ensure auto-deleted queues are properly cleaned up (i.e. are unbound from exchanges) to avoid leaking memory as messages are accumulated in inaccessible queues. (some cleanup to follow on this) git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@585417 13f79535-47bb-0310-9956-ffa450edef68 --- cpp/src/tests/QueueTest.cpp | 42 ++++++++++++++++++++++-------------------- 1 file changed, 22 insertions(+), 20 deletions(-) (limited to 'cpp/src/tests/QueueTest.cpp') diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp index f2f1b3bf84..114e0045f5 100644 --- a/cpp/src/tests/QueueTest.cpp +++ b/cpp/src/tests/QueueTest.cpp @@ -36,6 +36,8 @@ using namespace qpid::sys; class TestConsumer : public virtual Consumer{ public: + typedef shared_ptr shared_ptr; + Message::shared_ptr last; bool received; TestConsumer(): received(false) {}; @@ -85,8 +87,8 @@ class QueueTest : public CppUnit::TestCase Queue::shared_ptr queue(new Queue("my_test_queue", true)); Message::shared_ptr received; - TestConsumer c1; - queue->consume(&c1); + TestConsumer::shared_ptr c1(new TestConsumer()); + queue->consume(c1); //Test basic delivery: @@ -95,7 +97,7 @@ class QueueTest : public CppUnit::TestCase queue->process(msg1); sleep(2); - CPPUNIT_ASSERT(!c1.received); + CPPUNIT_ASSERT(!c1->received); msg1->enqueueComplete(); received = queue->dequeue().payload; @@ -124,10 +126,10 @@ class QueueTest : public CppUnit::TestCase Queue::shared_ptr queue(new Queue("my_queue", true)); //Test adding consumers: - TestConsumer c1; - TestConsumer c2; - queue->consume(&c1); - queue->consume(&c2); + TestConsumer::shared_ptr c1(new TestConsumer()); + TestConsumer::shared_ptr c2(new TestConsumer()); + queue->consume(c1); + queue->consume(c2); CPPUNIT_ASSERT_EQUAL(uint32_t(2), queue->getConsumerCount()); @@ -137,25 +139,25 @@ class QueueTest : public CppUnit::TestCase Message::shared_ptr msg3 = message("e", "C"); queue->deliver(msg1); - if (!c1.received) + if (!c1->received) sleep(2); - CPPUNIT_ASSERT_EQUAL(msg1.get(), c1.last.get()); + CPPUNIT_ASSERT_EQUAL(msg1.get(), c1->last.get()); queue->deliver(msg2); - if (!c2.received) + if (!c2->received) sleep(2); - CPPUNIT_ASSERT_EQUAL(msg2.get(), c2.last.get()); + CPPUNIT_ASSERT_EQUAL(msg2.get(), c2->last.get()); - c1.received = false; + c1->received = false; queue->deliver(msg3); - if (!c1.received) + if (!c1->received) sleep(2); - CPPUNIT_ASSERT_EQUAL(msg3.get(), c1.last.get()); + CPPUNIT_ASSERT_EQUAL(msg3.get(), c1->last.get()); //Test cancellation: - queue->cancel(&c1); + queue->cancel(c1); CPPUNIT_ASSERT_EQUAL(uint32_t(1), queue->getConsumerCount()); - queue->cancel(&c2); + queue->cancel(c2); CPPUNIT_ASSERT_EQUAL(uint32_t(0), queue->getConsumerCount()); } @@ -200,13 +202,13 @@ class QueueTest : public CppUnit::TestCase CPPUNIT_ASSERT_EQUAL(msg2.get(), received.get()); CPPUNIT_ASSERT_EQUAL(uint32_t(1), queue->getMessageCount()); - TestConsumer consumer; - queue->consume(&consumer); + TestConsumer::shared_ptr consumer(new TestConsumer()); + queue->consume(consumer); queue->requestDispatch(); - if (!consumer.received) + if (!consumer->received) sleep(2); - CPPUNIT_ASSERT_EQUAL(msg3.get(), consumer.last.get()); + CPPUNIT_ASSERT_EQUAL(msg3.get(), consumer->last.get()); CPPUNIT_ASSERT_EQUAL(uint32_t(0), queue->getMessageCount()); received = queue->dequeue().payload; -- cgit v1.2.1