Update from trunk r1375509 through r1450773asyncstore

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/asyncstore@1451244 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 19 insertions, 7 deletions

diff --git a/cpp/src/qpid/ha/HaPlugin.cpp b/cpp/src/qpid/ha/HaPlugin.cpp
index f7fe553d9b..d26b466847 100644
--- a/cpp/src/qpid/ha/HaPlugin.cpp
+++ b/cpp/src/qpid/ha/HaPlugin.cpp
@@ -33,9 +33,11 @@ struct Options : public qpid::Options {
         addOptions()
             ("ha-cluster", optValue(settings.cluster, "yes|no"),
              "Join a HA active/passive cluster.")
+            ("ha-queue-replication", optValue(settings.queueReplication, "yes|no"),
+             "Enable replication of specific queues without joining a cluster")
             ("ha-brokers-url", optValue(settings.brokerUrl,"URL"),
              "URL with address of each broker in the cluster.")
-            ("ha-public-url", optValue(settings.clientUrl,"URL"),
+            ("ha-public-url", optValue(settings.publicUrl,"URL"),
              "URL advertized to clients to connect to the cluster.")
             ("ha-replicate",
              optValue(settings.replicateDefault, "LEVEL"),
@@ -48,6 +50,10 @@ struct Options : public qpid::Options {
              "Authentication mechanism for connections between HA brokers")
             ("ha-backup-timeout", optValue(settings.backupTimeout, "SECONDS"),
              "Maximum time to wait for an expected backup to connect and become ready.")
+            ("ha-flow-messages", optValue(settings.flowMessages, "N"),
+             "Flow control message count limit for replication, 0 means no limit")
+            ("ha-flow-bytes", optValue(settings.flowBytes, "N"),
+             "Flow control byte limit for replication, 0 means no limit")
             ;
     }
 };
@@ -64,17 +70,23 @@ struct HaPlugin : public Plugin {
 
     void earlyInitialize(Plugin::Target& target) {
         broker::Broker* broker = dynamic_cast<broker::Broker*>(&target);
-        if (broker) {
-            // Must create the HaBroker in earlyInitialize so it can set up its
-            // connection observer before clients start connecting.
-            haBroker.reset(new ha::HaBroker(*broker, settings));
-            broker->addFinalizer(boost::bind(&HaPlugin::finalize, this));
+        if (broker && (settings.cluster || settings.queueReplication)) {
+            if (!broker->getManagementAgent()) {
+                QPID_LOG(info, "HA plugin disabled because management is disabled");
+                if (settings.cluster)
+                    throw Exception("Cannot start HA: management is disabled");
+            } else {
+                // Must create the HaBroker in earlyInitialize so it can set up its
+                // connection observer before clients start connecting.
+                haBroker.reset(new ha::HaBroker(*broker, settings));
+                broker->addFinalizer(boost::bind(&HaPlugin::finalize, this));
+            }
         }
     }
 
     void initialize(Plugin::Target& target) {
         broker::Broker* broker = dynamic_cast<broker::Broker*>(&target);
-        if (broker) haBroker->initialize();
+        if (broker && haBroker.get()) haBroker->initialize();
     }
 
     void finalize() {