Cluster support for copying shared broker state to new members.

cluster/DumpClient: Copies broker shared state to a new broker via AMQP.
broker/*Registry, Queue, QueueBindings: Added iteration functions for DumpClient
broker/SemanticState.cpp: Allow DumpClient to sidestep setting of delivery-properties.exchange.
client/Connection.h: Added Connection::open(Url) overload.
client/SessionImpl: Added send(AMQBody, FrameSet) overload for forwarding broker messages.
tests/cluster_test.cpp: Added test for DumpClient copying shared state between brokers.


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

1 files changed, 7 insertions, 6 deletions

diff --git a/cpp/src/qpid/broker/SemanticState.cpp b/cpp/src/qpid/broker/SemanticState.cpp
index 3185080f94..675a6a304c 100644
--- a/cpp/src/qpid/broker/SemanticState.cpp
+++ b/cpp/src/qpid/broker/SemanticState.cpp
@@ -353,8 +353,14 @@ void SemanticState::handle(intrusive_ptr<Message> msg) {
 void SemanticState::route(intrusive_ptr<Message> msg, Deliverable& strategy) {
     std::string exchangeName = msg->getExchangeName();
     //TODO: the following should be hidden behind message (using MessageAdapter or similar)
+
+    // Do not replace the delivery-properties.exchange if it is is already set.
+    // This is used internally (by the cluster) to force the exchange name on a message.
+    // The client library ensures this is always empty for messages from normal clients.
     if (msg->isA<MessageTransferBody>()) {
-        msg->getProperties<DeliveryProperties>()->setExchange(exchangeName);
+        if (!msg->hasProperties<DeliveryProperties>() ||
+            msg->getProperties<DeliveryProperties>()->getExchange().empty()) 
+            msg->getProperties<DeliveryProperties>()->setExchange(exchangeName);
     }
     if (!cacheExchange || cacheExchange->getName() != exchangeName){
         cacheExchange = session.getBroker().getExchanges().get(exchangeName);
@@ -392,11 +398,6 @@ void SemanticState::requestDispatch(ConsumerImpl& c)
 {    
     if(c.isBlocked())
         outputTasks.activateOutput();
-    // TODO aconway 2008-07-16:  we could directly call
-    //  c.doOutput();
-    // since we are in the connections thread but for consistency
-    // activateOutput() will set it up to be called in the next write idle.
-    // Current cluster code depends on this, review cluster code to change.
 }
 
 void SemanticState::complete(DeliveryRecord& delivery)