QPID-3931: Remove update exchanges once update is complete.

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

1 files changed, 16 insertions, 10 deletions

diff --git a/qpid/cpp/src/qpid/cluster/Cluster.cpp b/qpid/cpp/src/qpid/cluster/Cluster.cpp
index 3c1d23c842..799e9fc858 100644
--- a/qpid/cpp/src/qpid/cluster/Cluster.cpp
+++ b/qpid/cpp/src/qpid/cluster/Cluster.cpp
@@ -131,6 +131,7 @@
 #include "qpid/cluster/UpdateExchange.h"
 #include "qpid/cluster/ClusterTimer.h"
 #include "qpid/cluster/CredentialsExchange.h"
+#include "qpid/cluster/UpdateClient.h"
 
 #include "qpid/assert.h"
 #include "qmf/org/apache/qpid/cluster/ArgsClusterStopClusterNode.h"
@@ -269,7 +270,6 @@ Cluster::Cluster(const ClusterSettings& set, broker::Broker& b) :
                       "Error delivering frames",
                       poller),
     failoverExchange(new FailoverExchange(broker.GetVhostObject(), &broker)),
-    updateDataExchange(new UpdateDataExchange(*this)),
     credentialsExchange(new CredentialsExchange(*this)),
     quorum(boost::bind(&Cluster::leave, this)),
     decoder(boost::bind(&Cluster::deliverFrame, this, _1)),
@@ -295,15 +295,6 @@ Cluster::Cluster(const ClusterSettings& set, broker::Broker& b) :
     // Failover exchange provides membership updates to clients.
     broker.getExchanges().registerExchange(failoverExchange);
 
-    // Update exchange is used during updates to replicate messages
-    // without modifying delivery-properties.exchange.
-    broker.getExchanges().registerExchange(
-        boost::shared_ptr<broker::Exchange>(new UpdateExchange(this)));
-
-    // Update-data exchange is used for passing data that may be too large
-    // for single control frame.
-    broker.getExchanges().registerExchange(updateDataExchange);
-
     // CredentialsExchange is used to authenticate new cluster members
     broker.getExchanges().registerExchange(credentialsExchange);
 
@@ -680,6 +671,17 @@ void Cluster::initMapCompleted(Lock& l) {
             authenticate();
             broker.setRecovery(false); // Ditch my current store.
             broker.setClusterUpdatee(true);
+
+            // Update exchange is used during updates to replicate messages
+            // without modifying delivery-properties.exchange.
+            broker.getExchanges().registerExchange(
+                boost::shared_ptr<broker::Exchange>(new UpdateExchange(this)));
+
+            // Update-data exchange is used during update for passing data that
+            // may be too large for single control frame.
+            updateDataExchange.reset(new UpdateDataExchange(*this));
+            broker.getExchanges().registerExchange(updateDataExchange);
+
             if (mAgent) mAgent->suppress(true); // Suppress mgmt output during update.
             state = JOINER;
             mcast.mcastControl(ClusterUpdateRequestBody(ProtocolVersion(), myUrl.str()), self);
@@ -999,6 +1001,10 @@ void Cluster::checkUpdateIn(Lock& l) {
                         boost::ref(broker.getExchanges())));
         enableClusterSafe();    // Enable cluster-safe assertions
         deliverEventQueue.start();
+        // FIXME aconway 2012-04-04: unregister/delete Update[Data]Exchange
+        updateDataExchange.reset();
+        broker.getExchanges().destroy(UpdateDataExchange::EXCHANGE_NAME);
+        broker.getExchanges().destroy(UpdateClient::UPDATE);
     }
     else if (updateRetracted) { // Update was retracted, request another update
         updateRetracted = false;