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

1 files changed, 22 insertions, 20 deletions

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<TestConsumer> 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;