Pass QueuedMessage to queues consumers. This records the position of that message in the queue which is need to handle rlease and acquire.

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

4 files changed, 19 insertions, 17 deletions

diff --git a/cpp/src/tests/BrokerChannelTest.cpp b/cpp/src/tests/BrokerChannelTest.cpp
index 1e5a30f157..0787405eb7 100644
--- a/cpp/src/tests/BrokerChannelTest.cpp
+++ b/cpp/src/tests/BrokerChannelTest.cpp
@@ -256,13 +256,13 @@ class BrokerChannelTest : public CppUnit::TestCase
         queue->deliver(msg3);
 	sleep(2);
         
-        Message::shared_ptr next = queue->dequeue();
+        Message::shared_ptr next = queue->dequeue().payload;
         CPPUNIT_ASSERT_EQUAL(msg1, next);
         CPPUNIT_ASSERT_EQUAL((uint32_t) data1.size(), next->encodedContentSize());
-        next = queue->dequeue();
+        next = queue->dequeue().payload;
         CPPUNIT_ASSERT_EQUAL(msg2, next);
         CPPUNIT_ASSERT_EQUAL((uint32_t) data2.size(), next->encodedContentSize());
-        next = queue->dequeue();
+        next = queue->dequeue().payload;
         CPPUNIT_ASSERT_EQUAL(msg3, next);
         CPPUNIT_ASSERT_EQUAL((uint32_t) 0, next->encodedContentSize());
 
@@ -295,11 +295,11 @@ class BrokerChannelTest : public CppUnit::TestCase
         queue3->deliver(msg1);
 	sleep(2);
         
-        Message::shared_ptr next = queue1->dequeue();
+        Message::shared_ptr next = queue1->dequeue().payload;
         CPPUNIT_ASSERT_EQUAL(msg1, next);
-        next = queue2->dequeue();
+        next = queue2->dequeue().payload;
         CPPUNIT_ASSERT_EQUAL(msg1, next);
-        next = queue3->dequeue();
+        next = queue3->dequeue().payload;
         CPPUNIT_ASSERT_EQUAL(msg1, next);
 
         }
diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index ef1518af4c..bf742f9511 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -40,10 +40,10 @@ public:
     bool received;
     TestConsumer(): received(false) {};
 
-    virtual bool deliver(Message::shared_ptr& msg){
-    last = msg;
-    received = true;
-    return true;
+    virtual bool deliver(QueuedMessage& msg){
+        last = msg.payload;
+        received = true;
+        return true;
     };
 };
 
@@ -97,7 +97,7 @@ class QueueTest : public CppUnit::TestCase
         CPPUNIT_ASSERT(!c1.received);
  	msg1->enqueueComplete();
 
-        received = queue->dequeue();
+        received = queue->dequeue().payload;
         CPPUNIT_ASSERT_EQUAL(msg1.get(), received.get());
  	
    
@@ -190,11 +190,11 @@ class QueueTest : public CppUnit::TestCase
 
         CPPUNIT_ASSERT_EQUAL(uint32_t(3), queue->getMessageCount());
         
-        received = queue->dequeue();
+        received = queue->dequeue().payload;
         CPPUNIT_ASSERT_EQUAL(msg1.get(), received.get());
         CPPUNIT_ASSERT_EQUAL(uint32_t(2), queue->getMessageCount());
 
-        received = queue->dequeue();
+        received = queue->dequeue().payload;
         CPPUNIT_ASSERT_EQUAL(msg2.get(), received.get());
         CPPUNIT_ASSERT_EQUAL(uint32_t(1), queue->getMessageCount());
 
@@ -207,7 +207,7 @@ class QueueTest : public CppUnit::TestCase
         CPPUNIT_ASSERT_EQUAL(msg3.get(), consumer.last.get());
         CPPUNIT_ASSERT_EQUAL(uint32_t(0), queue->getMessageCount());
 
-        received = queue->dequeue();
+        received = queue->dequeue().payload;
         CPPUNIT_ASSERT(!received);
         CPPUNIT_ASSERT_EQUAL(uint32_t(0), queue->getMessageCount());
         
diff --git a/cpp/src/tests/TxAckTest.cpp b/cpp/src/tests/TxAckTest.cpp
index 89a907d495..65426e4e21 100644
--- a/cpp/src/tests/TxAckTest.cpp
+++ b/cpp/src/tests/TxAckTest.cpp
@@ -76,7 +76,9 @@ public:
             msg->getProperties<DeliveryProperties>()->setDeliveryMode(PERSISTENT);
             msg->getProperties<DeliveryProperties>()->setRoutingKey("routing_key");
             messages.push_back(msg);
-            deliveries.push_back(DeliveryRecord(msg, queue, "xyz", (i+1)));
+            QueuedMessage qm;
+            qm.payload = msg;
+            deliveries.push_back(DeliveryRecord(qm, queue, "xyz", (i+1)));
         }
 
         //assume msgs 1-5, 7 and 9 are all acked (i.e. 6, 8 & 10 are not)
diff --git a/cpp/src/tests/TxPublishTest.cpp b/cpp/src/tests/TxPublishTest.cpp
index 5628cf1d1c..4ec526f207 100644
--- a/cpp/src/tests/TxPublishTest.cpp
+++ b/cpp/src/tests/TxPublishTest.cpp
@@ -99,13 +99,13 @@ public:
         op.prepare(0);
         op.commit();
         CPPUNIT_ASSERT_EQUAL((uint32_t) 1, queue1->getMessageCount());
-	Message::shared_ptr msg_dequeue = queue1->dequeue();
+	Message::shared_ptr msg_dequeue = queue1->dequeue().payload;
 
  	CPPUNIT_ASSERT_EQUAL( true, ((PersistableMessage*) msg_dequeue.get())->isEnqueueComplete());
         CPPUNIT_ASSERT_EQUAL(msg, msg_dequeue);
 
         CPPUNIT_ASSERT_EQUAL((uint32_t) 1, queue2->getMessageCount());
-        CPPUNIT_ASSERT_EQUAL(msg, queue2->dequeue());            
+        CPPUNIT_ASSERT_EQUAL(msg, queue2->dequeue().payload);            
     }
 };