QPID-3767: fix federation and cluster failover support

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

4 files changed, 18 insertions, 9 deletions

diff --git a/qpid/cpp/src/qpid/broker/Link.cpp b/qpid/cpp/src/qpid/broker/Link.cpp
index 1cc723a717..b605ca71e5 100644
--- a/qpid/cpp/src/qpid/broker/Link.cpp
+++ b/qpid/cpp/src/qpid/broker/Link.cpp
@@ -269,7 +269,7 @@ void Link::setUrl(const Url& u) {
 namespace {
     /** invoked when session used to subscribe to remote's amq.failover exchange detaches */
     void sessionDetached(Link *link) {
-        QPID_LOG(debug, "detached from 'amq.failover' for link: " << link->getName());
+        QPID_LOG(notice, "detached from 'amq.failover' for link: " << link->getName());
     }
 }
 
@@ -763,6 +763,14 @@ std::string Link::createName(const std::string& transport,
     return linkName.str();
 }
 
+
+bool Link::pendingConnection(const std::string& _host, uint16_t _port) const
+{
+    Mutex::ScopedLock mutex(lock);
+    return (isConnecting() && _port == port && _host == host);
+}
+
+
 const std::string Link::exchangeTypeName("qpid.LinkExchange");
 
 }} // namespace qpid::broker
diff --git a/qpid/cpp/src/qpid/broker/Link.h b/qpid/cpp/src/qpid/broker/Link.h
index 312c425c95..5b788bb947 100644
--- a/qpid/cpp/src/qpid/broker/Link.h
+++ b/qpid/cpp/src/qpid/broker/Link.h
@@ -113,6 +113,7 @@ class Link : public PersistableConfig, public management::Manageable {
     void closed(int, std::string);   // Called when connection goes away
     void notifyConnectionForced(const std::string text);
     void closeConnection(const std::string& reason);
+    bool pendingConnection(const std::string& host, uint16_t port) const;  // is Link trying to connect to this remote?
 
     friend class LinkRegistry; // to call established, opened, closed
 
diff --git a/qpid/cpp/src/qpid/broker/LinkRegistry.cpp b/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
index e855d59fc1..3cad2c40c9 100644
--- a/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
+++ b/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
@@ -68,7 +68,7 @@ LinkRegistry::LinkRegistry (Broker* _broker) :
 
 LinkRegistry::~LinkRegistry() {}
 
-/** find link by current remote address */
+/** find link by the *configured* remote address */
 boost::shared_ptr<Link> LinkRegistry::getLink(const std::string& host,
                                               uint16_t           port,
                                               const std::string& transport)
@@ -319,18 +319,16 @@ void LinkRegistry::notifyConnection(const std::string& key, Connection* c)
     {
         Mutex::ScopedLock locker(lock);
         for (LinkMap::iterator l = links.begin(); l != links.end(); ++l) {
-            if (l->second->isConnecting() &&
-                l->second->getHost() == host &&
-                l->second->getPort() == port) {
+            if (l->second->pendingConnection(host, port)) {
                 link = l->second;
                 connections[key] = link->getName();
+                link->established(c);
                 break;
             }
         }
     }
 
     if (link) {
-        link->established(c);
         c->setUserId(str(format("%1%@%2%") % link->getUsername() % realm));
     }
 }
diff --git a/qpid/cpp/src/qpid/broker/LinkRegistry.h b/qpid/cpp/src/qpid/broker/LinkRegistry.h
index 0ee4125192..5f79d9bb52 100644
--- a/qpid/cpp/src/qpid/broker/LinkRegistry.h
+++ b/qpid/cpp/src/qpid/broker/LinkRegistry.h
@@ -86,10 +86,12 @@ namespace broker {
         /** determine if Link exists */
         QPID_BROKER_EXTERN boost::shared_ptr<Link>
           getLink(const std::string& name);
+        /** host,port,transport will be matched against the configured values, which may
+            be different from the current values due to failover */
         QPID_BROKER_EXTERN boost::shared_ptr<Link>
-          getLink(const std::string& host,
-                  uint16_t           port,
-                  const std::string& transport = std::string());
+          getLink(const std::string& configHost,
+                  uint16_t           configPort,
+                  const std::string& configTransport = std::string());
 
         QPID_BROKER_EXTERN std::pair<Bridge::shared_ptr, bool>
         declare(const std::string& name,