- support for store to abort init / force

- there is an issue exiting via exception not related to this fix, but uncovered
by this fix - JIRA 671


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

7 files changed, 19 insertions, 9 deletions

diff --git a/qpid/cpp/src/qpid/broker/Broker.cpp b/qpid/cpp/src/qpid/broker/Broker.cpp
index 01f8a04545..45f5be9cad 100644
--- a/qpid/cpp/src/qpid/broker/Broker.cpp
+++ b/qpid/cpp/src/qpid/broker/Broker.cpp
@@ -63,6 +63,7 @@ Broker::Options::Options(const std::string& name) :
     stagingThreshold(5000000),
     storeDir("/var"),
     storeAsync(false),
+    storeForce(false),
     enableMgmt(0),
     mgmtPubInterval(10),
     ack(100)
@@ -84,6 +85,8 @@ Broker::Options::Options(const std::string& name) :
          "Store directory location for persistence.")
         ("store-async", optValue(storeAsync,"yes|no"),
          "Use async persistence storage - if store supports it, enable AIO 0-DIRECT.")
+        ("store-force", optValue(storeForce,"yes|no"),
+         "Force changing modes of store, will delete all existing data if mode is change. Be SHURE you want to do this")
         ("mgmt,m", optValue(enableMgmt,"yes|no"),
          "Enable Management")
         ("mgmt-pub-interval", optValue(mgmtPubInterval, "SECONDS"),
@@ -138,10 +141,14 @@ Broker::Broker(const Broker::Options& conf) :
         QPID_LOG(info, "Management not enabled");
 
     if(store.get()) {
-        store->init(conf.storeDir, conf.storeAsync);
-        RecoveryManagerImpl recoverer(queues, exchanges, dtxManager, 
+        if (!store->init(conf.storeDir, conf.storeAsync, conf.storeForce)){
+              throw Exception( "Existing Journal in different mode, backup/move existing data \
+			  before changing modes. Or use --store-force yes to blow existing data away.");
+		}else{
+             RecoveryManagerImpl recoverer(queues, exchanges, dtxManager, 
                                       conf.stagingThreshold);
-        store->recover(recoverer);
+             store->recover(recoverer);
+        }
     }
 
     // Initialize plugins
diff --git a/qpid/cpp/src/qpid/broker/Broker.h b/qpid/cpp/src/qpid/broker/Broker.h
index 5a67fb7212..9ae6a5c8af 100644
--- a/qpid/cpp/src/qpid/broker/Broker.h
+++ b/qpid/cpp/src/qpid/broker/Broker.h
@@ -69,6 +69,7 @@ class Broker : public sys::Runnable, public Plugin::Target
         long stagingThreshold;
         string storeDir;
         bool storeAsync;
+        bool storeForce;
         bool enableMgmt;
         uint16_t mgmtPubInterval;
         uint32_t ack;
diff --git a/qpid/cpp/src/qpid/broker/MessageStore.h b/qpid/cpp/src/qpid/broker/MessageStore.h
index c8b243ef28..b88abc277a 100644
--- a/qpid/cpp/src/qpid/broker/MessageStore.h
+++ b/qpid/cpp/src/qpid/broker/MessageStore.h
@@ -44,8 +44,9 @@ public:
      * 
      * @param dir the directory to create logs/db's
      * @param async true, enable async, false, enable sync
+     * @param force true, delete data on mode change, false, error on mode change
 	 */
-	virtual void init(const std::string& dir, const bool async) = 0;
+	virtual bool init(const std::string& dir, const bool async, const bool force) = 0;
 
     /**
      * Record the existence of a durable queue
diff --git a/qpid/cpp/src/qpid/broker/MessageStoreModule.cpp b/qpid/cpp/src/qpid/broker/MessageStoreModule.cpp
index 384dfff6dc..1ddad65c56 100644
--- a/qpid/cpp/src/qpid/broker/MessageStoreModule.cpp
+++ b/qpid/cpp/src/qpid/broker/MessageStoreModule.cpp
@@ -28,9 +28,10 @@ MessageStoreModule::MessageStoreModule(const std::string& name) : store(name)
 {
 }
 
-void MessageStoreModule::init(const std::string& dir, const bool async)
+bool MessageStoreModule::init(const std::string& dir, const bool async, const bool force)
 {
-	store->init(dir, async);
+	store->init(dir, async, force);
+	return true;
 }
 
 void MessageStoreModule::create(PersistableQueue& queue)
diff --git a/qpid/cpp/src/qpid/broker/MessageStoreModule.h b/qpid/cpp/src/qpid/broker/MessageStoreModule.h
index 43772bf98c..1d256b972b 100644
--- a/qpid/cpp/src/qpid/broker/MessageStoreModule.h
+++ b/qpid/cpp/src/qpid/broker/MessageStoreModule.h
@@ -38,7 +38,7 @@ class MessageStoreModule : public MessageStore
 public:
     MessageStoreModule(const std::string& name);
 
-	void init(const std::string& dir, const bool async);
+	bool init(const std::string& dir, const bool async, const bool force);
     std::auto_ptr<TransactionContext> begin();
     std::auto_ptr<TPCTransactionContext> begin(const std::string& xid);
     void prepare(TPCTransactionContext& txn);
diff --git a/qpid/cpp/src/qpid/broker/NullMessageStore.cpp b/qpid/cpp/src/qpid/broker/NullMessageStore.cpp
index b3fc7ec9a7..2190ff4086 100644
--- a/qpid/cpp/src/qpid/broker/NullMessageStore.cpp
+++ b/qpid/cpp/src/qpid/broker/NullMessageStore.cpp
@@ -49,7 +49,7 @@ using namespace qpid::broker;
 
 NullMessageStore::NullMessageStore(bool _warn) : warn(_warn){}
 
-void NullMessageStore::init(const std::string& /*dir*/, const bool /*async*/) {}
+bool NullMessageStore::init(const std::string& /*dir*/, const bool /*async*/, const bool /*force*/) {return true;}
 
 void NullMessageStore::create(PersistableQueue& queue)
 {
diff --git a/qpid/cpp/src/qpid/broker/NullMessageStore.h b/qpid/cpp/src/qpid/broker/NullMessageStore.h
index 77723c0918..96aa286b82 100644
--- a/qpid/cpp/src/qpid/broker/NullMessageStore.h
+++ b/qpid/cpp/src/qpid/broker/NullMessageStore.h
@@ -38,7 +38,7 @@ class NullMessageStore : public MessageStore
 public:
     NullMessageStore(bool warn = false);
 
-	virtual void init(const std::string& dir, const bool async);
+	virtual bool init(const std::string& dir, const bool async, const bool force);
     virtual std::auto_ptr<TransactionContext> begin();
     virtual std::auto_ptr<TPCTransactionContext> begin(const std::string& xid);
     virtual void prepare(TPCTransactionContext& txn);