Added some methods to MessageStore interface and hooked these in where appropriate.

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

5 files changed, 146 insertions, 138 deletions

diff --git a/cpp/test/unit/qpid/broker/ConfigurationTest.cpp b/cpp/test/unit/qpid/broker/ConfigurationTest.cpp
index 7acee7c8b9..8fd252a9a9 100644
--- a/cpp/test/unit/qpid/broker/ConfigurationTest.cpp
+++ b/cpp/test/unit/qpid/broker/ConfigurationTest.cpp
@@ -60,8 +60,8 @@ class ConfigurationTest : public CppUnit::TestCase
     void testVarious() 
     {
         Configuration conf;
-        char* argv[] = {"ignore", "-t", "--worker-threads", "10", "-a", "blocking"};
-        conf.parse(6, argv);
+        char* argv[] = {"ignore", "-t", "--worker-threads", "10"};
+        conf.parse(4, argv);
         CPPUNIT_ASSERT_EQUAL(5672, conf.getPort());//default
         CPPUNIT_ASSERT_EQUAL(10, conf.getWorkerThreads());
         CPPUNIT_ASSERT(conf.isTrace());
diff --git a/cpp/test/unit/qpid/broker/ExchangeTest.cpp b/cpp/test/unit/qpid/broker/ExchangeTest.cpp
index 2fb525312b..14fb6fc097 100644
--- a/cpp/test/unit/qpid/broker/ExchangeTest.cpp
+++ b/cpp/test/unit/qpid/broker/ExchangeTest.cpp
@@ -37,8 +37,8 @@ class ExchangeTest : public CppUnit::TestCase
 
     void testMe() 
     {
-        Queue::shared_ptr queue(new Queue("queue", true, true));
-        Queue::shared_ptr queue2(new Queue("queue2", true, true));
+        Queue::shared_ptr queue(new Queue("queue", true));
+        Queue::shared_ptr queue2(new Queue("queue2", true));
 
         TopicExchange topic("topic");
         topic.bind(queue, "abc", 0);
diff --git a/cpp/test/unit/qpid/broker/QueueTest.cpp b/cpp/test/unit/qpid/broker/QueueTest.cpp
index 5b06cb93ca..ba1427a087 100644
--- a/cpp/test/unit/qpid/broker/QueueTest.cpp
+++ b/cpp/test/unit/qpid/broker/QueueTest.cpp
@@ -52,7 +52,7 @@ class QueueTest : public CppUnit::TestCase
 
   public:
     void testConsumers(){
-        Queue::shared_ptr queue(new Queue("my_queue", true, true));
+        Queue::shared_ptr queue(new Queue("my_queue", true));
     
         //Test adding consumers:
         TestConsumer c1; 
@@ -84,7 +84,7 @@ class QueueTest : public CppUnit::TestCase
     }
 
     void testBinding(){
-        Queue::shared_ptr queue(new Queue("my_queue", true, true));
+        Queue::shared_ptr queue(new Queue("my_queue", true));
         //Test bindings:
         TestBinding a;
         TestBinding b;
@@ -118,7 +118,7 @@ class QueueTest : public CppUnit::TestCase
     }
 
     void testDequeue(){
-        Queue::shared_ptr queue(new Queue("my_queue", true, true));
+        Queue::shared_ptr queue(new Queue("my_queue", true));
 
         Message::shared_ptr msg1 = Message::shared_ptr(new Message(0, "e", "A", true, true));
         Message::shared_ptr msg2 = Message::shared_ptr(new Message(0, "e", "B", true, true));
diff --git a/cpp/test/unit/qpid/broker/TxAckTest.cpp b/cpp/test/unit/qpid/broker/TxAckTest.cpp
index b787c5793b..ab1e607e87 100644
--- a/cpp/test/unit/qpid/broker/TxAckTest.cpp
+++ b/cpp/test/unit/qpid/broker/TxAckTest.cpp
@@ -16,6 +16,7 @@
  *
  */
 #include "qpid/broker/MessageStore.h"
+#include "qpid/broker/QueueRegistry.h"
 #include "qpid/broker/TxAck.h"
 #include <qpid_test_plugin.h>
 #include <iostream>
@@ -29,81 +30,84 @@ using namespace qpid::broker;
 class TxAckTest : public CppUnit::TestCase  
 {
 
-        class TestMessageStore : public MessageStore
-        {
-            public:
-                vector<Message::shared_ptr> dequeued;
-
-                void dequeue(Message::shared_ptr& msg, const string& /*queue*/, const string * const /*xid*/)
-                {
-                    dequeued.push_back(msg);
-                }
-
-                //dont care about any of the other methods:
-                void enqueue(Message::shared_ptr&, const string&, const string * const){}
-                void committed(const string * const){}
-                void aborted(const string * const){}
-                void begin(){}
-                void commit(){}
-                void abort(){}        
-                ~TestMessageStore(){}
-        };
-
-        CPPUNIT_TEST_SUITE(TxAckTest);
-        CPPUNIT_TEST(testPrepare);
-        CPPUNIT_TEST(testCommit);
-        CPPUNIT_TEST_SUITE_END();
-
-
-        AccumulatedAck acked;
-        TestMessageStore store;
-        Queue::shared_ptr queue;
-        vector<Message::shared_ptr> messages;
-        list<DeliveryRecord> deliveries;
-        TxAck op;
-
-
+    class TestMessageStore : public MessageStore
+    {
     public:
+        vector<Message::shared_ptr> dequeued;
 
-        TxAckTest() : queue(new Queue("my_queue", true, false, &store, 0)), op(acked, deliveries)
+        void dequeue(Message::shared_ptr& msg, const Queue& /*queue*/, const string * const /*xid*/)
         {
-            for(int i = 0; i < 10; i++){
-                Message::shared_ptr msg(new Message(0, "exchange", "routing_key", false, false));
-                messages.push_back(msg);
-                deliveries.push_back(DeliveryRecord(msg, queue, "xyz", (i+1)));
-            }
-
-            //assume msgs 1-5, 7 and 9 are all acked (i.e. 6, 8 & 10 are not)
-            acked.range = 5;
-            acked.individual.push_back(7);
-            acked.individual.push_back(9);
-        }      
-
-        void testPrepare()
-        {
-            //ensure acked messages are discarded, i.e. dequeued from store
-            op.prepare();
-            CPPUNIT_ASSERT_EQUAL((size_t) 7, store.dequeued.size());
-            CPPUNIT_ASSERT_EQUAL((size_t) 10, deliveries.size());
-            CPPUNIT_ASSERT_EQUAL(messages[0], store.dequeued[0]);//msg 1
-            CPPUNIT_ASSERT_EQUAL(messages[1], store.dequeued[1]);//msg 2
-            CPPUNIT_ASSERT_EQUAL(messages[2], store.dequeued[2]);//msg 3
-            CPPUNIT_ASSERT_EQUAL(messages[3], store.dequeued[3]);//msg 4
-            CPPUNIT_ASSERT_EQUAL(messages[4], store.dequeued[4]);//msg 5
-            CPPUNIT_ASSERT_EQUAL(messages[6], store.dequeued[5]);//msg 7
-            CPPUNIT_ASSERT_EQUAL(messages[8], store.dequeued[6]);//msg 9
+            dequeued.push_back(msg);
         }
 
-        void testCommit()
-        {
-            //emsure acked messages are removed from list
-            op.commit();
-            CPPUNIT_ASSERT_EQUAL((size_t) 3, deliveries.size());
-            list<DeliveryRecord>::iterator i = deliveries.begin();
-            CPPUNIT_ASSERT(i->matches(6));//msg 6
-            CPPUNIT_ASSERT((++i)->matches(8));//msg 8
-            CPPUNIT_ASSERT((++i)->matches(10));//msg 10
+        //dont care about any of the other methods:
+        void create(const Queue&){}
+        void destroy(const Queue&){}        
+        void recover(QueueRegistry&){}
+        void enqueue(Message::shared_ptr&, const Queue&, const string * const){}
+        void committed(const string * const){}
+        void aborted(const string * const){}
+        void begin(){}
+        void commit(){}
+        void abort(){}        
+        ~TestMessageStore(){}
+    };
+
+    CPPUNIT_TEST_SUITE(TxAckTest);
+    CPPUNIT_TEST(testPrepare);
+    CPPUNIT_TEST(testCommit);
+    CPPUNIT_TEST_SUITE_END();
+
+
+    AccumulatedAck acked;
+    TestMessageStore store;
+    Queue::shared_ptr queue;
+    vector<Message::shared_ptr> messages;
+    list<DeliveryRecord> deliveries;
+    TxAck op;
+
+
+public:
+
+    TxAckTest() : queue(new Queue("my_queue", false, &store, 0)), op(acked, deliveries)
+    {
+        for(int i = 0; i < 10; i++){
+            Message::shared_ptr msg(new Message(0, "exchange", "routing_key", false, false));
+            messages.push_back(msg);
+            deliveries.push_back(DeliveryRecord(msg, queue, "xyz", (i+1)));
         }
+
+        //assume msgs 1-5, 7 and 9 are all acked (i.e. 6, 8 & 10 are not)
+        acked.range = 5;
+        acked.individual.push_back(7);
+        acked.individual.push_back(9);
+    }      
+
+    void testPrepare()
+    {
+        //ensure acked messages are discarded, i.e. dequeued from store
+        op.prepare();
+        CPPUNIT_ASSERT_EQUAL((size_t) 7, store.dequeued.size());
+        CPPUNIT_ASSERT_EQUAL((size_t) 10, deliveries.size());
+        CPPUNIT_ASSERT_EQUAL(messages[0], store.dequeued[0]);//msg 1
+        CPPUNIT_ASSERT_EQUAL(messages[1], store.dequeued[1]);//msg 2
+        CPPUNIT_ASSERT_EQUAL(messages[2], store.dequeued[2]);//msg 3
+        CPPUNIT_ASSERT_EQUAL(messages[3], store.dequeued[3]);//msg 4
+        CPPUNIT_ASSERT_EQUAL(messages[4], store.dequeued[4]);//msg 5
+        CPPUNIT_ASSERT_EQUAL(messages[6], store.dequeued[5]);//msg 7
+        CPPUNIT_ASSERT_EQUAL(messages[8], store.dequeued[6]);//msg 9
+    }
+
+    void testCommit()
+    {
+        //emsure acked messages are removed from list
+        op.commit();
+        CPPUNIT_ASSERT_EQUAL((size_t) 3, deliveries.size());
+        list<DeliveryRecord>::iterator i = deliveries.begin();
+        CPPUNIT_ASSERT(i->matches(6));//msg 6
+        CPPUNIT_ASSERT((++i)->matches(8));//msg 8
+        CPPUNIT_ASSERT((++i)->matches(10));//msg 10
+    }
 };
 
 // Make this test suite a plugin.
diff --git a/cpp/test/unit/qpid/broker/TxPublishTest.cpp b/cpp/test/unit/qpid/broker/TxPublishTest.cpp
index b8d3c99cb9..4fe6c7497a 100644
--- a/cpp/test/unit/qpid/broker/TxPublishTest.cpp
+++ b/cpp/test/unit/qpid/broker/TxPublishTest.cpp
@@ -16,6 +16,7 @@
  *
  */
 #include "qpid/broker/MessageStore.h"
+#include "qpid/broker/QueueRegistry.h"
 #include "qpid/broker/TxPublish.h"
 #include <qpid_test_plugin.h>
 #include <iostream>
@@ -30,71 +31,74 @@ using namespace qpid::broker;
 class TxPublishTest : public CppUnit::TestCase  
 {
 
-        class TestMessageStore : public MessageStore
-        {
-            public:
-                vector< pair<string, Message::shared_ptr> > enqueued;
-
-                void enqueue(Message::shared_ptr& msg, const string& queue, const string * const /*xid*/)
-                {
-                    enqueued.push_back(pair<string, Message::shared_ptr>(queue,msg));
-                }
-
-                //dont care about any of the other methods:
-                void dequeue(Message::shared_ptr&, const string&, const string * const){}
-                void committed(const string * const){}
-                void aborted(const string * const){}
-                void begin(){}
-                void commit(){}
-                void abort(){}        
-                ~TestMessageStore(){}
-        };
-
-        CPPUNIT_TEST_SUITE(TxPublishTest);
-        CPPUNIT_TEST(testPrepare);
-        CPPUNIT_TEST(testCommit);
-        CPPUNIT_TEST_SUITE_END();
-
-
-        TestMessageStore store;
-        Queue::shared_ptr queue1;
-        Queue::shared_ptr queue2;
-        Message::shared_ptr msg;
-        TxPublish op;
-
-
+    class TestMessageStore : public MessageStore
+    {
     public:
-
-        TxPublishTest() : queue1(new Queue("queue1", true, false, &store, 0)), 
-                          queue2(new Queue("queue2", true, false, &store, 0)), 
-                          msg(new Message(0, "exchange", "routing_key", false, false)),
-                          op(msg)
-        {
-            op.deliverTo(queue1);
-            op.deliverTo(queue2);
-        }      
-
-        void testPrepare()
+        vector< pair<string, Message::shared_ptr> > enqueued;
+        
+        void enqueue(Message::shared_ptr& msg, const Queue& queue, const string * const /*xid*/)
         {
-            //ensure messages are enqueued in store
-            op.prepare();
-            CPPUNIT_ASSERT_EQUAL((size_t) 2, store.enqueued.size());
-            CPPUNIT_ASSERT_EQUAL(string("queue1"), store.enqueued[0].first);
-            CPPUNIT_ASSERT_EQUAL(msg, store.enqueued[0].second);
-            CPPUNIT_ASSERT_EQUAL(string("queue2"), store.enqueued[1].first);
-            CPPUNIT_ASSERT_EQUAL(msg, store.enqueued[1].second);
-        }
-
-        void testCommit()
-        {
-            //ensure messages are delivered to queue
-            op.commit();
-            CPPUNIT_ASSERT_EQUAL((u_int32_t) 1, queue1->getMessageCount());
-            CPPUNIT_ASSERT_EQUAL(msg, queue1->dequeue());
-
-            CPPUNIT_ASSERT_EQUAL((u_int32_t) 1, queue2->getMessageCount());
-            CPPUNIT_ASSERT_EQUAL(msg, queue2->dequeue());            
+            enqueued.push_back(pair<string, Message::shared_ptr>(queue.getName(),msg));
         }
+        
+        //dont care about any of the other methods:
+        void create(const Queue&){}
+        void destroy(const Queue&){}
+        void recover(QueueRegistry&){}
+        void dequeue(Message::shared_ptr&, const Queue&, const string * const){}
+        void committed(const string * const){}
+        void aborted(const string * const){}
+        void begin(){}
+        void commit(){}
+        void abort(){}        
+        ~TestMessageStore(){}
+    };
+    
+    CPPUNIT_TEST_SUITE(TxPublishTest);
+    CPPUNIT_TEST(testPrepare);
+    CPPUNIT_TEST(testCommit);
+    CPPUNIT_TEST_SUITE_END();
+    
+    
+    TestMessageStore store;
+    Queue::shared_ptr queue1;
+    Queue::shared_ptr queue2;
+    Message::shared_ptr msg;
+    TxPublish op;
+    
+    
+public:
+    
+    TxPublishTest() : queue1(new Queue("queue1", false, &store, 0)), 
+                      queue2(new Queue("queue2", false, &store, 0)), 
+                      msg(new Message(0, "exchange", "routing_key", false, false)),
+                      op(msg)
+    {
+        op.deliverTo(queue1);
+        op.deliverTo(queue2);
+    }      
+
+    void testPrepare()
+    {
+        //ensure messages are enqueued in store
+        op.prepare();
+        CPPUNIT_ASSERT_EQUAL((size_t) 2, store.enqueued.size());
+        CPPUNIT_ASSERT_EQUAL(string("queue1"), store.enqueued[0].first);
+        CPPUNIT_ASSERT_EQUAL(msg, store.enqueued[0].second);
+        CPPUNIT_ASSERT_EQUAL(string("queue2"), store.enqueued[1].first);
+        CPPUNIT_ASSERT_EQUAL(msg, store.enqueued[1].second);
+    }
+
+    void testCommit()
+    {
+        //ensure messages are delivered to queue
+        op.commit();
+        CPPUNIT_ASSERT_EQUAL((u_int32_t) 1, queue1->getMessageCount());
+        CPPUNIT_ASSERT_EQUAL(msg, queue1->dequeue());
+
+        CPPUNIT_ASSERT_EQUAL((u_int32_t) 1, queue2->getMessageCount());
+        CPPUNIT_ASSERT_EQUAL(msg, queue2->dequeue());            
+    }
 };
 
 // Make this test suite a plugin.