- async message fix in dequeue

- addition of test for async enqueue of messages on BrokerQueue



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

2 files changed, 27 insertions, 1 deletions

diff --git a/cpp/src/qpid/broker/BrokerQueue.cpp b/cpp/src/qpid/broker/BrokerQueue.cpp
index d67575103f..e2b12ef316 100644
--- a/cpp/src/qpid/broker/BrokerQueue.cpp
+++ b/cpp/src/qpid/broker/BrokerQueue.cpp
@@ -189,7 +189,8 @@ Message::shared_ptr Queue::dequeue(){
     Message::shared_ptr msg;
     if(!messages.empty()){
         msg = messages.front();
-        pop();
+	if (msg->isEnqueueComplete())
+           pop();
     }
     return msg;
 }
diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index fad2702f38..6b30f9ff42 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -63,6 +63,7 @@ class QueueTest : public CppUnit::TestCase
     CPPUNIT_TEST(testRegistry);
     CPPUNIT_TEST(testDequeue);
     CPPUNIT_TEST(testBound);
+    CPPUNIT_TEST(testAsyncMessage);
     CPPUNIT_TEST_SUITE_END();
 
 
@@ -72,6 +73,30 @@ class QueueTest : public CppUnit::TestCase
             new BasicMessage(0, exchange, routingKey, false, false));
     }
     
+    
+    void testAsyncMessage(){
+    
+        Queue::shared_ptr queue(new Queue("my_test_queue", true));
+        Message::shared_ptr received;
+	
+        TestConsumer c1; 
+        queue->consume(&c1);
+
+       
+        //Test basic delivery:
+        Message::shared_ptr msg1 = message("e", "A");
+        queue->process(msg1);
+	sleep(2);
+
+        CPPUNIT_ASSERT(!c1.received);
+ 	msg1->enqueueComplete();
+
+        received = queue->dequeue();
+        CPPUNIT_ASSERT_EQUAL(msg1.get(), received.get());
+ 	
+   
+    }
+    
     void testConsumers(){
         Queue::shared_ptr queue(new Queue("my_queue", true));