QPID-4156: HA close window for clients to connect before HA broker is initialized; merged r1363498

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

3 files changed, 31 insertions, 10 deletions

diff --git a/qpid/cpp/src/qpid/ha/HaBroker.cpp b/qpid/cpp/src/qpid/ha/HaBroker.cpp
index a0502b94ae..14b5218198 100644
--- a/qpid/cpp/src/qpid/ha/HaBroker.cpp
+++ b/qpid/cpp/src/qpid/ha/HaBroker.cpp
@@ -55,6 +55,7 @@ using types::Variant;
 using types::Uuid;
 using sys::Mutex;
 
+// Called in Plugin::earlyInitialize
 HaBroker::HaBroker(broker::Broker& b, const Settings& s)
     : logPrefix("Broker: "),
       broker(b),
@@ -63,12 +64,26 @@ HaBroker::HaBroker(broker::Broker& b, const Settings& s)
       observer(new ConnectionObserver(*this, systemId)),
       mgmtObject(0),
       status(STANDALONE),
-      brokerInfo(broker.getSystem()->getNodeName(),
-                 // TODO aconway 2012-05-24: other transports?
-                 broker.getPort(broker::Broker::TCP_TRANSPORT), systemId),
       membership(systemId),
       replicationTest(s.replicateDefault.get())
 {
+    // If we are joining a cluster we must start excluding clients now,
+    // otherwise there's a window for a client to connect before we get to
+    // initialize()
+    if (settings.cluster) {
+        observer->setObserver(boost::shared_ptr<broker::ConnectionObserver>(
+                              new BackupConnectionExcluder));
+        broker.getConnectionObservers().add(observer);
+    }
+}
+
+// Called in Plugin::initialize
+void HaBroker::initialize() {
+
+    // FIXME aconway 2012-07-19: assumes there's a TCP transport with a meaningful port.
+    brokerInfo = BrokerInfo(
+        broker.getSystem()->getNodeName(), broker.getPort(broker::Broker::TCP_TRANSPORT), systemId);
+
     // Set up the management object.
     ManagementAgent* ma = broker.getManagementAgent();
     if (settings.cluster && !ma)
@@ -87,10 +102,7 @@ HaBroker::HaBroker(broker::Broker& b, const Settings& s)
     // If we are in a cluster, start as backup in joining state.
     if (settings.cluster) {
         status = JOINING;
-        observer->setObserver(boost::shared_ptr<broker::ConnectionObserver>(
-                                  new BackupConnectionExcluder));
-        broker.getConnectionObservers().add(observer);
-        backup.reset(new Backup(*this, s));
+        backup.reset(new Backup(*this, settings));
         broker.getKnownBrokers = boost::bind(&HaBroker::getKnownBrokers, this);
     }
 
diff --git a/qpid/cpp/src/qpid/ha/HaBroker.h b/qpid/cpp/src/qpid/ha/HaBroker.h
index 7daba018cf..7e47cadf7e 100644
--- a/qpid/cpp/src/qpid/ha/HaBroker.h
+++ b/qpid/cpp/src/qpid/ha/HaBroker.h
@@ -63,9 +63,13 @@ class Primary;
 class HaBroker : public management::Manageable
 {
   public:
+    /** HaBroker is constructed during earlyInitialize */
     HaBroker(broker::Broker&, const Settings&);
     ~HaBroker();
 
+    /** Called during plugin initialization */
+    void initialize();
+
     // Implement Manageable.
     qpid::management::ManagementObject* GetManagementObject() const { return mgmtObject; }
     management::Manageable::status_t ManagementMethod (
diff --git a/qpid/cpp/src/qpid/ha/HaPlugin.cpp b/qpid/cpp/src/qpid/ha/HaPlugin.cpp
index 360b6892ab..73f4ed5e8c 100644
--- a/qpid/cpp/src/qpid/ha/HaPlugin.cpp
+++ b/qpid/cpp/src/qpid/ha/HaPlugin.cpp
@@ -62,16 +62,21 @@ struct HaPlugin : public Plugin {
 
     Options* getOptions() { return &options; }
 
-    void earlyInitialize(Plugin::Target& ) {}
-
-    void initialize(Plugin::Target& target) {
+    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 conneting.
             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();
+    }
+
     void finalize() {
         haBroker.reset();
     }