Replicate connection decoder fragments to new members.

Refactoring:
 - Merge Decoder into ConnectionMap.
 - Process cluster controls in event queue thread.
 - Use counter not pointer for connection ID, avoid re-use.
 - Do all processing in event queue thread to avoid races
   (temporary pending performance measurements)


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

1 files changed, 7 insertions, 8 deletions

diff --git a/cpp/src/qpid/cluster/UpdateClient.cpp b/cpp/src/qpid/cluster/UpdateClient.cpp
index 18746ccb7e..4a4af4adbd 100644
--- a/cpp/src/qpid/cluster/UpdateClient.cpp
+++ b/cpp/src/qpid/cluster/UpdateClient.cpp
@@ -95,7 +95,7 @@ UpdateClient::UpdateClient(const MemberId& updater, const MemberId& updatee, con
     : updaterId(updater), updateeId(updatee), updateeUrl(url), updaterBroker(broker), map(m),
       frameId(frameId_), connections(cons), 
       connection(catchUpConnection()), shadowConnection(catchUpConnection()),
-      done(ok), failed(fail) 
+      done(ok), failed(fail), connectionSettings(cs)
 {
     connection.open(url, cs);
     session = connection.newSession("update_shared");
@@ -228,13 +228,15 @@ void UpdateClient::updateConnection(const boost::intrusive_ptr<Connection>& upda
     shadowConnection = catchUpConnection();
 
     broker::Connection& bc = updateConnection->getBrokerConnection();
-    // FIXME aconway 2008-10-20: What authentication info to use on reconnect?
-    shadowConnection.open(updateeUrl, bc.getUserId(), ""/*password*/, "/"/*vhost*/, bc.getFrameMax());
+    connectionSettings.maxFrameSize = bc.getFrameMax();
+    shadowConnection.open(updateeUrl, connectionSettings);
     bc.eachSessionHandler(boost::bind(&UpdateClient::updateSession, this, _1));
+    std::pair<const char*, size_t> fragment = updateConnection->getDecoder().getFragment();
     ClusterConnectionProxy(shadowConnection).shadowReady(
         updateConnection->getId().getMember(),
-        reinterpret_cast<uint64_t>(updateConnection->getId().getPointer()),
-        updateConnection->getBrokerConnection().getUserId()
+        updateConnection->getId().getNumber(),
+        bc.getUserId(),
+        string(fragment.first, fragment.second)
     );
     shadowConnection.close();
     QPID_LOG(debug, updaterId << " updated connection " << *updateConnection);
@@ -285,9 +287,6 @@ void UpdateClient::updateSession(broker::SessionHandler& sh) {
     if (inProgress) {
         inProgress->getFrames().map(simpl->out);
     }
-
-    // FIXME aconway 2008-09-23: update session replay list.
-
     QPID_LOG(debug, updaterId << " updated session " << sh.getSession()->getId());
 }