- added init for dir and async options for store

- added flush for queue for async processing



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

6 files changed, 46 insertions, 2 deletions

diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index 92eb6dd380..f21109f818 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -103,7 +103,8 @@ Broker::Broker(const Broker::Options& conf) :
     exchanges.declare(amq_match, HeadersExchange::typeName);
 
     if(store.get()) {
-        RecoveryManagerImpl recoverer(queues, exchanges, dtxManager, 
+		store->init(conf.storeDir, conf.storeAsync);
+		RecoveryManagerImpl recoverer(queues, exchanges, dtxManager, 
                                       conf.stagingThreshold);
         store->recover(recoverer);
     }
diff --git a/cpp/src/qpid/broker/MessageStore.h b/cpp/src/qpid/broker/MessageStore.h
index 6bd661954f..c8b243ef28 100644
--- a/cpp/src/qpid/broker/MessageStore.h
+++ b/cpp/src/qpid/broker/MessageStore.h
@@ -37,6 +37,16 @@ namespace broker {
  */
 class MessageStore : public TransactionalStore, public Recoverable {
 public:
+
+    /**
+     * init the store, call before any other call. If not called, store 
+     * is free to pick any defaults
+     * 
+     * @param dir the directory to create logs/db's
+     * @param async true, enable async, false, enable sync
+	 */
+	virtual void init(const std::string& dir, const bool async) = 0;
+
     /**
      * Record the existence of a durable queue
      */
@@ -133,6 +143,15 @@ public:
      */
     virtual void dequeue(TransactionContext* ctxt, PersistableMessage& msg, const PersistableQueue& queue) = 0;
 
+    /**
+     * Flushes all async messages to disk for the specified queue
+     *
+     * Note: that this is async so the return of the function does
+     * not mean the opperation is complete.
+     * 
+     * @param queue the name of the queue from which it is to be dequeued
+     */
+    virtual void flush(const qpid::broker::PersistableQueue& queue)=0;
 
    /**
      * Returns the number of outstanding AIO's for a given queue
diff --git a/cpp/src/qpid/broker/MessageStoreModule.cpp b/cpp/src/qpid/broker/MessageStoreModule.cpp
index c87503279e..384dfff6dc 100644
--- a/cpp/src/qpid/broker/MessageStoreModule.cpp
+++ b/cpp/src/qpid/broker/MessageStoreModule.cpp
@@ -28,6 +28,11 @@ MessageStoreModule::MessageStoreModule(const std::string& name) : store(name)
 {
 }
 
+void MessageStoreModule::init(const std::string& dir, const bool async)
+{
+	store->init(dir, async);
+}
+
 void MessageStoreModule::create(PersistableQueue& queue)
 {
     store->create(queue);
@@ -95,6 +100,11 @@ void MessageStoreModule::dequeue(TransactionContext* ctxt, PersistableMessage& m
     store->dequeue(ctxt, msg, queue);
 }
 
+void MessageStoreModule::flush(const qpid::broker::PersistableQueue& queue)
+{
+    store->flush(queue);
+}
+
 u_int32_t MessageStoreModule::outstandingQueueAIO(const PersistableQueue& queue)
 {
     return store->outstandingQueueAIO(queue);
diff --git a/cpp/src/qpid/broker/MessageStoreModule.h b/cpp/src/qpid/broker/MessageStoreModule.h
index a983957b5e..43772bf98c 100644
--- a/cpp/src/qpid/broker/MessageStoreModule.h
+++ b/cpp/src/qpid/broker/MessageStoreModule.h
@@ -38,6 +38,7 @@ class MessageStoreModule : public MessageStore
 public:
     MessageStoreModule(const std::string& name);
 
+	void init(const std::string& dir, const bool async);
     std::auto_ptr<TransactionContext> begin();
     std::auto_ptr<TPCTransactionContext> begin(const std::string& xid);
     void prepare(TPCTransactionContext& txn);
@@ -62,7 +63,8 @@ public:
     void enqueue(TransactionContext* ctxt, PersistableMessage& msg, const PersistableQueue& queue);
     void dequeue(TransactionContext* ctxt, PersistableMessage& msg, const PersistableQueue& queue);
     u_int32_t outstandingQueueAIO(const PersistableQueue& queue);
-
+    void flush(const qpid::broker::PersistableQueue& queue);
+	 
     ~MessageStoreModule(){}
 };
 
diff --git a/cpp/src/qpid/broker/NullMessageStore.cpp b/cpp/src/qpid/broker/NullMessageStore.cpp
index 843d31654b..b8e4950287 100644
--- a/cpp/src/qpid/broker/NullMessageStore.cpp
+++ b/cpp/src/qpid/broker/NullMessageStore.cpp
@@ -49,6 +49,11 @@ using namespace qpid::broker;
 
 NullMessageStore::NullMessageStore(bool _warn) : warn(_warn){}
 
+void NullMessageStore::init(const std::string& /*dir*/, const bool /*async*/)
+{
+    QPID_LOG(info, "Can't init, store not enabled");
+}
+
 void NullMessageStore::create(PersistableQueue& queue)
 {
     QPID_LOG(info, "Can't create durable queue '" << queue.getName() << "'. Persistence not enabled.");
@@ -109,6 +114,11 @@ void NullMessageStore::dequeue(TransactionContext*, PersistableMessage& msg, con
     QPID_LOG(info, "Can't dequeue message from '" << queue.getName() << "'. Persistence not enabled.");
 }
 
+void NullMessageStore::flush(const qpid::broker::PersistableQueue& queue)
+{
+    QPID_LOG(info, "Can't flush. Persistence not enabled queue-" << queue.getName());
+}
+
 u_int32_t NullMessageStore::outstandingQueueAIO(const PersistableQueue& )
 {
     return 0;
diff --git a/cpp/src/qpid/broker/NullMessageStore.h b/cpp/src/qpid/broker/NullMessageStore.h
index bb7b245717..77723c0918 100644
--- a/cpp/src/qpid/broker/NullMessageStore.h
+++ b/cpp/src/qpid/broker/NullMessageStore.h
@@ -38,6 +38,7 @@ class NullMessageStore : public MessageStore
 public:
     NullMessageStore(bool warn = false);
 
+	virtual void init(const std::string& dir, const bool async);
     virtual std::auto_ptr<TransactionContext> begin();
     virtual std::auto_ptr<TPCTransactionContext> begin(const std::string& xid);
     virtual void prepare(TPCTransactionContext& txn);
@@ -62,6 +63,7 @@ public:
     virtual void enqueue(TransactionContext* ctxt, PersistableMessage& msg, const PersistableQueue& queue);
     virtual void dequeue(TransactionContext* ctxt, PersistableMessage& msg, const PersistableQueue& queue);
     virtual u_int32_t outstandingQueueAIO(const PersistableQueue& queue);
+	virtual void flush(const qpid::broker::PersistableQueue& queue);
     ~NullMessageStore(){}
 };