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

6 files changed, 107 insertions, 9 deletions

diff --git a/cpp/src/qpid/client/Connection.cpp b/cpp/src/qpid/client/Connection.cpp
index 6572794516..e19cfea059 100644
--- a/cpp/src/qpid/client/Connection.cpp
+++ b/cpp/src/qpid/client/Connection.cpp
@@ -23,6 +23,7 @@
 #include "Message.h"
 #include "SessionImpl.h"
 #include "SessionBase_0_10Access.h"
+#include "qpid/Url.h"
 #include "qpid/log/Logger.h"
 #include "qpid/log/Options.h"
 #include "qpid/log/Statement.h"
@@ -48,6 +49,37 @@ Connection::Connection() : channelIdCounter(0), version(framing::highestProtocol
 Connection::~Connection(){ }
 
 void Connection::open(
+    const Url& url,
+    const std::string& uid, const std::string& pwd, 
+    const std::string& vhost,
+    uint16_t maxFrameSize)
+{
+    if (url.empty())
+        throw Exception(QPID_MSG("Attempt to open URL with no addresses."));
+    Url::const_iterator i = url.begin();
+    do {
+        const TcpAddress* tcp = i->get<TcpAddress>();
+        i++;
+        if (tcp) {
+            try {
+                ConnectionSettings settings;
+                settings.host = tcp->host;
+                settings.port = tcp->port;
+                settings.username = uid;
+                settings.password = pwd;
+                settings.virtualhost = vhost;
+                settings.maxFrameSize = maxFrameSize;
+                open(settings);
+                break;
+            }
+            catch (const Exception& e) {
+                if (i == url.end()) throw;
+            }
+        }
+    } while (i != url.end());
+}
+
+void Connection::open(
     const std::string& host, int port,
     const std::string& uid, const std::string& pwd, 
     const std::string& vhost,
diff --git a/cpp/src/qpid/client/Connection.h b/cpp/src/qpid/client/Connection.h
index ee543e20d2..e8ff2d8660 100644
--- a/cpp/src/qpid/client/Connection.h
+++ b/cpp/src/qpid/client/Connection.h
@@ -1,5 +1,5 @@
-#ifndef _client_Connection_
-#define _client_Connection_
+#ifndef QPID_CLIENT_CONNECTION_H
+#define QPID_CLIENT_CONNECTION_H
 
 /*
  *
@@ -26,6 +26,9 @@
 #include "qpid/client/Session.h"
 
 namespace qpid {
+
+class Url;
+
 namespace client {
 
 class ConnectionSettings;
@@ -77,6 +80,28 @@ class Connection
               const std::string& virtualhost = "/", uint16_t maxFrameSize=65535);
 
     /**
+     * Opens a connection to a broker using a URL.
+     * If the URL contains multiple addresses, try each in turn
+     * till connection is successful.
+     * 
+     * @url address of the broker to connect to.
+     *
+     * @param uid the userid to connect with.
+     *
+     * @param pwd the password to connect with (currently SASL
+     * PLAIN is the only authentication method supported so this
+     * is sent in clear text).
+     * 
+     * @param virtualhost the AMQP virtual host to use (virtual
+     * hosts, where implemented(!), provide namespace partitioning
+     * within a single broker).
+     */
+    void open(const Url& url,
+              const std::string& uid = "guest",
+              const std::string& pwd = "guest", 
+              const std::string& virtualhost = "/", uint16_t maxFrameSize=65535);
+
+    /**
      * Opens a connection to a broker.
      * 
      * @param the settings to use (host, port etc) @see ConnectionSettings
@@ -146,4 +171,4 @@ class Connection
 }} // namespace qpid::client
 
 
-#endif
+#endif  /*!QPID_CLIENT_CONNECTION_H*/
diff --git a/cpp/src/qpid/client/Message.cpp b/cpp/src/qpid/client/Message.cpp
index d5464594ee..13caaecefd 100644
--- a/cpp/src/qpid/client/Message.cpp
+++ b/cpp/src/qpid/client/Message.cpp
@@ -24,9 +24,7 @@
 namespace qpid {
 namespace client {
 
-Message::Message(const std::string& data_,
-                 const std::string& routingKey,
-                 const std::string& exchange) : TransferContent(data_, routingKey, exchange) {}
+Message::Message(const std::string& data, const std::string& routingKey) : TransferContent(data, routingKey) {}
 
 std::string Message::getDestination() const 
 { 
diff --git a/cpp/src/qpid/client/Message.h b/cpp/src/qpid/client/Message.h
index 4e6ed49bb4..18efdd20f7 100644
--- a/cpp/src/qpid/client/Message.h
+++ b/cpp/src/qpid/client/Message.h
@@ -40,11 +40,9 @@ public:
     /** Create a Message.
      *@param data Data for the message body.
      *@param routingKey Passed to the exchange that routes the message.
-     *@param exchange Name of the exchange that should route the message.
      */
     Message(const std::string& data=std::string(),
-            const std::string& routingKey=std::string(),
-            const std::string& exchange=std::string());
+            const std::string& routingKey=std::string());
 
     /** The destination of messages sent to the broker is the exchange
      * name.  The destination of messages received from the broker is
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){
diff --git a/cpp/src/qpid/client/SessionImpl.h b/cpp/src/qpid/client/SessionImpl.h
index 55031a94ae..c63774a23a 100644
--- a/cpp/src/qpid/client/SessionImpl.h
+++ b/cpp/src/qpid/client/SessionImpl.h
@@ -81,6 +81,7 @@ public:
 
     Future send(const framing::AMQBody& command);
     Future send(const framing::AMQBody& command, const framing::MethodContent& content);
+    Future send(const framing::AMQBody& command, const framing::FrameSet& content);
 
     Demux& getDemux();
     void markCompleted(const framing::SequenceNumber& id, bool cumulative, bool notifyPeer);