QPID-3603: HA broker backup/primary ready checks.

- Introduce HA broker state machien
- Inform backup queues when ready.
- Incomplete implementation of backup ready check.
 - does not count correctly after a failover, see countUnready.
 - Existing replicator bridges updated out of sync with BrokerReplicator initialize.
 - Does not handle multi-messages responses.
- Newly promoted HA primary waits for backups to be ready before accepting clients.

- Uniform log prefixes for HA messages.
- qpid-ha tests, call qpid-ha python code directly.

- Move excluder from Backup to HaBroker, it is also used in PROMOTING.

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

1 files changed, 21 insertions, 4 deletions

diff --git a/cpp/src/qpid/ha/ConnectionExcluder.cpp b/cpp/src/qpid/ha/ConnectionExcluder.cpp
index 67ad7202d6..fef4c67174 100644
--- a/cpp/src/qpid/ha/ConnectionExcluder.cpp
+++ b/cpp/src/qpid/ha/ConnectionExcluder.cpp
@@ -27,14 +27,31 @@
 namespace qpid {
 namespace ha {
 
-ConnectionExcluder::ConnectionExcluder() {}
+ConnectionExcluder::ConnectionExcluder(const LogPrefix& lp)
+    : logPrefix(lp), backupAllowed(false) {}
 
 void ConnectionExcluder::opened(broker::Connection& connection) {
-    if (!connection.isLink() && !connection.getClientProperties().isSet(ADMIN_TAG))
-        throw Exception(
-            QPID_MSG("HA: Backup broker rejected connection " << connection.getMgmtId()));
+    if (connection.isLink()) return; // Allow all outgoing links
+    if (connection.getClientProperties().isSet(ADMIN_TAG)) {
+        QPID_LOG(debug, logPrefix << "Allowing admin connection: "
+                 << connection.getMgmtId());
+        return;
+    }
+    if (connection.getClientProperties().isSet(BACKUP_TAG)) {
+        if (backupAllowed) {
+            QPID_LOG(debug, logPrefix << "Allowing backup connection: "
+                     << connection.getMgmtId());
+            return;
+        }
+        else QPID_LOG(debug, logPrefix << "Rejected backup connection: "
+                      << connection.getMgmtId());
+    }
+
+    throw Exception(
+        QPID_MSG(logPrefix << "Rejected client connection " << connection.getMgmtId()));
 }
 
 const std::string ConnectionExcluder::ADMIN_TAG="qpid.ha-admin";
+const std::string ConnectionExcluder::BACKUP_TAG="qpid.ha-backup";
 
 }} // namespace qpid::ha