QPID-3603: HA broker close backup link when promoted.

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

4 files changed, 22 insertions, 13 deletions

diff --git a/qpid/cpp/src/qpid/broker/Link.cpp b/qpid/cpp/src/qpid/broker/Link.cpp
index e0c94853d7..2c053220a4 100644
--- a/qpid/cpp/src/qpid/broker/Link.cpp
+++ b/qpid/cpp/src/qpid/broker/Link.cpp
@@ -321,7 +321,7 @@ void Link::maintenanceVisit ()
         {
             visitCount = 0;
             //switch host and port to next in url list if possible
-            if (!tryFailover()) {
+            if (!tryFailoverLH()) {
                 currentInterval *= 2;
                 if (currentInterval > MAX_INTERVAL)
                     currentInterval = MAX_INTERVAL;
@@ -347,7 +347,7 @@ void Link::reconnect(const qpid::Address& a)
     }
 }
 
-bool Link::tryFailover() {      // FIXME aconway 2012-01-30: lock held?
+bool Link::tryFailoverLH() {      // FIXME aconway 2012-01-30: lock held?
     if (reconnectNext >= url.size()) reconnectNext = 0;
     if (url.empty()) return false;
     Address next = url[reconnectNext++];
@@ -440,18 +440,24 @@ ManagementObject* Link::GetManagementObject (void) const
     return (ManagementObject*) mgmtObject;
 }
 
+void Link::close() {
+    Mutex::ScopedLock mutex(lock);
+    if (!closing) {
+        closing = true;
+        if (state != STATE_CONNECTING && connection) {
+            //connection can only be closed on the connections own IO processing thread
+            connection->requestIOProcessing(boost::bind(&Link::destroy, this));
+        }
+    }
+}
+
+
 Manageable::status_t Link::ManagementMethod (uint32_t op, Args& args, string& text)
 {
     switch (op)
     {
     case _qmf::Link::METHOD_CLOSE :
-        if (!closing) {
-	    closing = true;
-	    if (state != STATE_CONNECTING && connection) {
-                //connection can only be closed on the connections own IO processing thread
-                connection->requestIOProcessing(boost::bind(&Link::destroy, this));
-	    }
-        }
+        close();
         return Manageable::STATUS_OK;
 
     case _qmf::Link::METHOD_BRIDGE :
diff --git a/qpid/cpp/src/qpid/broker/Link.h b/qpid/cpp/src/qpid/broker/Link.h
index a9d045b0af..9c82eda000 100644
--- a/qpid/cpp/src/qpid/broker/Link.h
+++ b/qpid/cpp/src/qpid/broker/Link.h
@@ -84,7 +84,7 @@ namespace qpid {
             void startConnectionLH();        // Start the IO Connection
             void destroy();                  // Called when mgmt deletes this link
             void ioThreadProcessing();       // Called on connection's IO thread by request
-            bool tryFailover();              // Called during maintenance visit
+            bool tryFailoverLH();            // Called during maintenance visit
             bool hideManagement() const;
 
         public:
@@ -119,6 +119,7 @@ namespace qpid {
             void closed(int, std::string);   // Called when connection goes away
             void setConnection(Connection*); // Set pointer to the AMQP Connection
             void reconnect(const Address&); //called by LinkRegistry
+            void close();       // Close the link from within the broker.
 
             std::string getAuthMechanism() { return authMechanism; }
             std::string getUsername()      { return username; }
diff --git a/qpid/cpp/src/qpid/ha/Backup.cpp b/qpid/cpp/src/qpid/ha/Backup.cpp
index e5bd0ed4dc..1ab24057ce 100644
--- a/qpid/cpp/src/qpid/ha/Backup.cpp
+++ b/qpid/cpp/src/qpid/ha/Backup.cpp
@@ -65,11 +65,12 @@ Backup::Backup(broker::Broker& b, const Settings& s) :
 }
 
 void Backup::setUrl(const Url& url) {
-    // FIXME aconway 2012-01-30: locking?
+    sys::Mutex::ScopedLock l(lock);
     link->setUrl(url);
 }
 
 Backup::~Backup() {
+    link->close();
     broker.getExchanges().destroy(replicator->getName());
     broker.getConnectionObservers().remove(excluder); // Allows client connections.
 }
diff --git a/qpid/cpp/src/qpid/ha/Backup.h b/qpid/cpp/src/qpid/ha/Backup.h
index 00ec55a6ff..3a9e739c98 100644
--- a/qpid/cpp/src/qpid/ha/Backup.h
+++ b/qpid/cpp/src/qpid/ha/Backup.h
@@ -24,6 +24,7 @@
 
 #include "Settings.h"
 #include "qpid/Url.h"
+#include "qpid/sys/Mutex.h"
 #include <boost/shared_ptr.hpp>
 
 namespace qpid {
@@ -41,8 +42,7 @@ class BrokerReplicator;
 /**
  * State associated with a backup broker. Manages connections to primary.
  *
- * THREAD SAFE: trivially because currently it only has a constructor.
- * May need locking as the functionality grows.
+ * THREAD SAFE
  */
 class Backup
 {
@@ -52,6 +52,7 @@ class Backup
     void setUrl(const Url&);
 
   private:
+    sys::Mutex lock;
     broker::Broker& broker;
     Settings settings;
     boost::shared_ptr<broker::Link> link;