QPID-1567: More changes to make clustering and federation work together

* replicate outgoing link traffic to all nodes

* coordinate amongst nodes so that only one node actually maintains active links
  with the others able to take over if that node fails



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

1 files changed, 13 insertions, 1 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 0a40493350..7b40328f1c 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -323,10 +323,20 @@ void Cluster::configChange(const MemberId&, const std::string& addresses, Lock&
             state = NEWBIE;
             QPID_LOG(info, *this << " joining cluster: " << map);
             mcast.mcastControl(ClusterDumpRequestBody(ProtocolVersion(), myUrl.str()), myId);
+            ClusterMap::Set members = map.getAlive();
+            members.erase(myId);
+            myElders = members;
+            broker.getLinks().setPassive(true);
         }
     }
-    else if (state >= READY && memberChange)
+    else if (state >= READY && memberChange) {
         memberUpdate(l);
+        myElders = ClusterMap::intersection(myElders, map.getAlive());
+        if (myElders.empty()) {
+            //assume we are oldest, reactive links if necessary
+            broker.getLinks().setPassive(false);
+        }
+    }
 }
 
 
@@ -496,6 +506,8 @@ void Cluster::memberUpdate(Lock& l) {
     }
     lastSize = size;
 
+    //
+
     if (mgmtObject) {
         mgmtObject->set_clusterSize(size); 
         string urlstr;