qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kgiu...@apache.org
Subject svn commit: r1332657 - in /qpid/trunk/qpid/cpp: src/qpid/broker/Link.cpp src/qpid/broker/Link.h src/qpid/cluster/Connection.cpp src/qpid/cluster/Connection.h src/qpid/cluster/UpdateClient.cpp xml/cluster.xml
Date Tue, 01 May 2012 13:57:50 GMT
Author: kgiusti
Date: Tue May  1 13:57:50 2012
New Revision: 1332657

URL: http://svn.apache.org/viewvc?rev=1332657&view=rev
Log:
QPID-3963: replicate learned failover addresses to new cluster members

Modified:
    qpid/trunk/qpid/cpp/src/qpid/broker/Link.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/Link.h
    qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
    qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
    qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
    qpid/trunk/qpid/cpp/xml/cluster.xml

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Link.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Link.cpp?rev=1332657&r1=1332656&r2=1332657&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Link.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Link.cpp Tue May  1 13:57:50 2012
@@ -104,7 +104,7 @@ public:
             urlVec = urlArrayToVector(addresses);
             for(size_t i = 0; i < urlVec.size(); ++i)
                 urls.insert(urls.end(), urlVec[i].begin(), urlVec[i].end());
-            QPID_LOG(notice, "Remote broker has provided these failover addresses= " <<
urls);
+            QPID_LOG(debug, "Remote broker has provided these failover addresses= " <<
urls);
             link->setUrl(urls);
         }
     }
@@ -253,6 +253,7 @@ void Link::established(Connection* c)
 
 
 void Link::setUrl(const Url& u) {
+    QPID_LOG(info, "Setting remote broker failover addresses for link '" << getName()
<< "' to these urls: " << u);
     Mutex::ScopedLock mutex(lock);
     url = u;
     reconnectNext = 0;
@@ -687,6 +688,35 @@ bool Link::getRemoteAddress(qpid::Addres
 }
 
 
+// FieldTable keys for internal state data
+namespace {
+    const std::string FAILOVER_ADDRESSES("failover-addresses");
+    const std::string FAILOVER_INDEX("failover-index");
+}
+
+void Link::getState(framing::FieldTable& state) const
+{
+    state.clear();
+    Mutex::ScopedLock mutex(lock);
+    if (!url.empty()) {
+        state.setString(FAILOVER_ADDRESSES, url.str());
+        state.setInt(FAILOVER_INDEX, reconnectNext);
+    }
+}
+
+void Link::setState(const framing::FieldTable& state)
+{
+    Mutex::ScopedLock mutex(lock);
+    if (state.isSet(FAILOVER_ADDRESSES)) {
+        Url failovers(state.getAsString(FAILOVER_ADDRESSES));
+        setUrl(failovers);
+    }
+    if (state.isSet(FAILOVER_INDEX)) {
+        reconnectNext = state.getAsInt(FAILOVER_INDEX);
+    }
+}
+
+
 const std::string Link::exchangeTypeName("qpid.LinkExchange");
 
 }} // namespace qpid::broker

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Link.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Link.h?rev=1332657&r1=1332656&r2=1332657&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Link.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Link.h Tue May  1 13:57:50 2012
@@ -51,7 +51,7 @@ class LinkExchange;
 
 class Link : public PersistableConfig, public management::Manageable {
   private:
-    sys::Mutex          lock;
+    mutable sys::Mutex  lock;
     LinkRegistry*       links;
     MessageStore*       store;
 
@@ -174,6 +174,10 @@ class Link : public PersistableConfig, p
     // manage the exchange owned by this link
     static const std::string exchangeTypeName;
     static boost::shared_ptr<Exchange> linkExchangeFactory(const std::string& name);
+
+    // replicate internal state of this Link for clustering
+    void getState(framing::FieldTable& state) const;
+    void setState(const framing::FieldTable& state);
 };
 }
 }

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp?rev=1332657&r1=1332656&r2=1332657&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp Tue May  1 13:57:50 2012
@@ -797,6 +797,54 @@ void Connection::config(const std::strin
     else throw Exception(QPID_MSG("Update failed, invalid kind of config: " << kind));
 }
 
+namespace {
+    // find a Link that matches the given Address
+    class LinkFinder {
+        qpid::Address id;
+        boost::shared_ptr<broker::Link> link;
+    public:
+        LinkFinder(const qpid::Address& _id) : id(_id) {}
+        boost::shared_ptr<broker::Link> getLink() { return link; }
+        void operator() (boost::shared_ptr<broker::Link> l)
+        {
+            if (!link) {
+                qpid::Address addr(l->getTransport(), l->getHost(), l->getPort());
+                if (id == addr) {
+                    link = l;
+                }
+            }
+        }
+    };
+}
+
+void Connection::internalState(const std::string& type,
+                               const std::string& name,
+                               const framing::FieldTable& state)
+{
+    if (type == "link") {
+        // name is the string representation of the Link's _configured_ destination address
+        Url dest;
+        try {
+            dest = name;
+        } catch(...) {
+            throw Exception(QPID_MSG("Update failed, invalid format for Link destination
address: " << name));
+        }
+        assert(dest.size());
+        LinkFinder finder(dest[0]);
+        cluster.getBroker().getLinks().eachLink(boost::ref(finder));
+        if (finder.getLink()) {
+            try {
+                finder.getLink()->setState(state);
+            } catch(...) {
+                throw Exception(QPID_MSG("Update failed, invalid state for Link " <<
name << ", state: " << state));
+            }
+            QPID_LOG(debug, cluster << " updated link " << dest[0] << "
with state: " << state);
+        } else throw Exception(QPID_MSG("Update failed, unable to find Link named: " <<
name));
+    }
+    else throw Exception(QPID_MSG("Update failed, invalid object type for internal state
replication: " << type));
+}
+
+
 void Connection::doCatchupIoCallbacks() {
     // We need to process IO callbacks during the catch-up phase in
     // order to service asynchronous completions for messages

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h?rev=1332657&r1=1332656&r2=1332657&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h Tue May  1 13:57:50 2012
@@ -200,6 +200,8 @@ class Connection :
                               const std::string& instance);
 
     void config(const std::string& encoded);
+    void internalState(const std::string& type, const std::string& name,
+                       const framing::FieldTable& state);
 
     void setSecureConnection ( broker::SecureConnection * sc );
 

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp?rev=1332657&r1=1332656&r2=1332657&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp Tue May  1 13:57:50 2012
@@ -688,7 +688,15 @@ void UpdateClient::updateLinks() {
 void UpdateClient::updateLink(const boost::shared_ptr<broker::Link>& link) {
     QPID_LOG(debug, *this << " updating link "
              << link->getHost() << ":" << link->getPort());
-    ClusterConnectionProxy(session).config(encode(*link));
+    ClusterConnectionProxy(session).config(encode(*link));  // push the configuration
+    // now push the current state
+    framing::FieldTable state;
+    link->getState(state);
+    std::ostringstream os;
+    os << qpid::Address(link->getTransport(), link->getHost(), link->getPort());
+    ClusterConnectionProxy(session).internalState(std::string("link"),
+                                                  os.str(),
+                                                  state);
 }
 
 void UpdateClient::updateBridge(const boost::shared_ptr<broker::Bridge>& bridge)
{

Modified: qpid/trunk/qpid/cpp/xml/cluster.xml
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/xml/cluster.xml?rev=1332657&r1=1332656&r2=1332657&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/xml/cluster.xml (original)
+++ qpid/trunk/qpid/cpp/xml/cluster.xml Tue May  1 13:57:50 2012
@@ -326,6 +326,12 @@
       <field name="dequeueSincePurge" type="uint32"/>
     </control>
 
+    <!-- Replicate the internal state for an object - e.g. Links, bridges, etc -->
+    <control name="internal-state" code="0x42">
+      <field name="type" type="str8"/> <!-- The type of object the state is for
(e.g. 'link') -->
+      <field name="name" type="str8"/> <!-- Identifies the particular object to
be updated -->
+      <field name="state" type="map"/> <!-- The internal state for the object -->
+    </control>
 
   </class>
 



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message