QPID-1306

Cleaner consume check, pointed out by gsim.



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

2 files changed, 12 insertions, 24 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index d50e887df4..735a116cf2 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -244,7 +244,7 @@ bool Queue::consumeNextMessage(QueuedMessage& m, Consumer::shared_ptr c)
             return false;
         } else {
             QueuedMessage msg = messages.front();
-            if (store && !msg.payload->isEnqueueComplete()) { 
+            if (!optimisticConsume && store && !msg.payload->isEnqueueComplete()) { 
                 QPID_LOG(debug, "Messages not ready to dispatch on queue '" << name << "'");
                 addListener(c);
                 return false;
@@ -482,11 +482,7 @@ bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg)
     }
 
     if (msg->isPersistent() && store) {
-        if (optimisticConsume){
-		    msg->enqueueComplete(); // (optimistic) allow consume before written to disk
-        } else {
-		    msg->enqueueAsync(shared_from_this(), store); //increment to async counter -- for message sent to more than one queue
-		}
+	    msg->enqueueAsync(shared_from_this(), store); //increment to async counter -- for message sent to more than one queue
 		boost::intrusive_ptr<PersistableMessage> pmsg = boost::static_pointer_cast<PersistableMessage>(msg);
         store->enqueue(ctxt, pmsg, *this);
         return true;
@@ -503,11 +499,7 @@ bool Queue::dequeue(TransactionContext* ctxt, const QueuedMessage& msg)
         dequeued(msg);
     }
     if (msg.payload->isPersistent() && store) {
-        if (optimisticConsume) {
-		    msg.payload->dequeueComplete();
-        } else {
-            msg.payload->dequeueAsync(shared_from_this(), store); //increment to async counter -- for message sent to more than one queue
-        }
+        msg.payload->dequeueAsync(shared_from_this(), store); //increment to async counter -- for message sent to more than one queue
 		boost::intrusive_ptr<PersistableMessage> pmsg = boost::static_pointer_cast<PersistableMessage>(msg.payload);
         store->dequeue(ctxt, pmsg, *this);
         return true;
diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index 111920aa59..ae29d48111 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -301,27 +301,23 @@ QPID_AUTO_TEST_CASE(testOptimisticConsume){
 	queue->setLastNodeFailure();
 	
     intrusive_ptr<Message> msg1 = message("e", "A");
-    intrusive_ptr<Message> msg2 = message("e", "B");
-    intrusive_ptr<Message> msg3 = message("e", "C");
 	msg1->forcePersistent();
-	msg2->forcePersistent();
-	msg3->forcePersistent();
 
-	//enqueue 2 messages
-    queue->deliver(msg1);
-    queue->deliver(msg2);
-	
 	//change mode
 	args.setInt("qpid.optimistic_consume", 1);
     queue->configure(args);
 	
 	//enqueue 1 message
-    queue->deliver(msg3);
+    queue->deliver(msg1);
 	
-	//check all have persistent ids.
-    BOOST_CHECK(!msg1->isEnqueueComplete());
-    BOOST_CHECK(!msg2->isEnqueueComplete());
-    BOOST_CHECK(msg3->isEnqueueComplete());
+    TestConsumer::shared_ptr consumer(new TestConsumer());
+    queue->consume(consumer);
+    queue->dispatch(consumer);
+    if (!consumer->received)
+        sleep(2);
+
+    BOOST_CHECK_EQUAL(msg1.get(), consumer->last.get());
+    BOOST_CHECK_EQUAL(uint32_t(0), queue->getMessageCount());
 
 }