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, 44 insertions, 0 deletions

diff --git a/cpp/src/qpid/client/SessionImpl.cpp b/cpp/src/qpid/client/SessionImpl.cpp
index aef546c427..68955050b4 100644
--- a/cpp/src/qpid/client/SessionImpl.cpp
+++ b/cpp/src/qpid/client/SessionImpl.cpp
@@ -28,9 +28,11 @@
 #include "qpid/framing/ClientInvoker.h"
 #include "qpid/framing/enum.h"
 #include "qpid/framing/FrameSet.h"
+#include "qpid/framing/AMQFrame.h"
 #include "qpid/framing/MethodContent.h"
 #include "qpid/framing/SequenceSet.h"
 #include "qpid/framing/reply_exceptions.h"
+#include "qpid/framing/DeliveryProperties.h"
 #include "qpid/log/Statement.h"
 
 #include <boost/bind.hpp>
@@ -272,6 +274,41 @@ Future SessionImpl::send(const AMQBody& command, const MethodContent& content)
     return sendCommand(command, &content);
 }
 
+namespace {
+// Functor for FrameSet::map to send header + content frames but, not method frames.
+struct SendContentFn {
+    FrameHandler& handler;
+    void operator()(const AMQFrame& f) {
+        if (!f.getMethod()) 
+            handler(const_cast<AMQFrame&>(f));
+    }
+    SendContentFn(FrameHandler& h) : handler(h) {}
+};
+}
+    
+Future SessionImpl::send(const AMQBody& command, const FrameSet& content) {
+    Acquire a(sendLock);
+    SequenceNumber id = nextOut++;
+    {
+        Lock l(state);
+        checkOpen();    
+        incompleteOut.add(id);
+    }
+    Future f(id);
+    if (command.getMethod()->resultExpected()) {        
+        Lock l(state);
+        //result listener must be set before the command is sent
+        f.setFutureResult(results.listenForResult(id));
+    }
+    AMQFrame frame(command);
+    frame.setEof(false);
+    handleOut(frame);
+
+    SendContentFn send(out);
+    content.map(send);
+    return f;
+}
+
 Future SessionImpl::sendCommand(const AMQBody& command, const MethodContent* content)
 {
     Acquire a(sendLock);
@@ -297,9 +334,16 @@ Future SessionImpl::sendCommand(const AMQBody& command, const MethodContent* con
     }
     return f;
 }
+
 void SessionImpl::sendContent(const MethodContent& content)
 {
     AMQFrame header(content.getHeader());
+
+    // Client is not allowed to set the delivery-properties.exchange.
+    AMQHeaderBody* headerp = static_cast<AMQHeaderBody*>(header.getBody());
+    if (headerp && headerp->get<DeliveryProperties>())
+        headerp->get<DeliveryProperties>(true)->clearExchangeFlag();
+
     header.setFirstSegment(false);
     uint64_t data_length = content.getData().length();
     if(data_length > 0){