QPID-1306

This patch includes:
- Optimistic Consume
- Support for forcing Queue durable on cluster failure
- Some cleanup on mgnt functions in Queue to inlines
- Tests

Still coming
- header for client queue options
- LVQ support bits.




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

1 files changed, 90 insertions, 0 deletions

diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index 8795dbcd03..111920aa59 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -24,6 +24,7 @@
 #include "qpid/broker/Deliverable.h"
 #include "qpid/broker/ExchangeRegistry.h"
 #include "qpid/broker/QueueRegistry.h"
+#include "qpid/broker/NullMessageStore.h"
 #include "qpid/framing/MessageTransferBody.h"
 #include <iostream>
 #include "boost/format.hpp"
@@ -236,6 +237,95 @@ QPID_AUTO_TEST_CASE(testBound)
     exchange3->route(deliverable, key, &args);
 }
 
+QPID_AUTO_TEST_CASE(testPersistLastNodeStanding){
+
+    FieldTable args;
+
+    // set queue mode
+	args.setInt("qpid.persist_last_node", 1);
+	
+	Queue::shared_ptr queue(new Queue("my-queue", true));
+    queue->configure(args);
+	
+    intrusive_ptr<Message> msg1 = message("e", "A");
+    intrusive_ptr<Message> msg2 = message("e", "B");
+    intrusive_ptr<Message> msg3 = message("e", "C");
+
+	//enqueue 2 messages
+    queue->deliver(msg1);
+    queue->deliver(msg2);
+	
+	//change mode
+	queue->setLastNodeFailure();
+	
+	//enqueue 1 message
+    queue->deliver(msg3);
+	
+	//check all have persistent ids.
+    BOOST_CHECK(msg1->isPersistent());
+    BOOST_CHECK(msg2->isPersistent());
+    BOOST_CHECK(msg3->isPersistent());
+
+}
+
+class TestMessageStore : public NullMessageStore
+{
+  public:
+    
+    virtual void dequeue(TransactionContext*,
+                 const boost::intrusive_ptr<PersistableMessage>& /*msg*/,
+                 const PersistableQueue& /*queue*/)
+    {
+    }
+
+    virtual void enqueue(TransactionContext*,
+                 const boost::intrusive_ptr<PersistableMessage>& /*msg*/,
+                 const PersistableQueue& /* queue */)
+    {
+    }
+
+    TestMessageStore() : NullMessageStore(false) {}
+    ~TestMessageStore(){}
+};
+
+
+QPID_AUTO_TEST_CASE(testOptimisticConsume){
+
+    FieldTable args;
+	args.setInt("qpid.persist_last_node", 1);
+
+    // set queue mode
+	
+	TestMessageStore store;
+	Queue::shared_ptr queue(new Queue("my-queue", true, &store));
+	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);
+	
+	//check all have persistent ids.
+    BOOST_CHECK(!msg1->isEnqueueComplete());
+    BOOST_CHECK(!msg2->isEnqueueComplete());
+    BOOST_CHECK(msg3->isEnqueueComplete());
+
+}
+
+
 QPID_AUTO_TEST_SUITE_END()