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, 211 insertions, 6 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 4c0a768c4f..ce156e85e4 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -25,7 +25,7 @@
 #include "qpid/framing/AMQFrame.h"
 #include "qpid/framing/AMQP_AllOperations.h"
 #include "qpid/framing/AllInvoker.h"
-#include "qpid/framing/ClusterUrlNoticeBody.h"
+#include "qpid/framing/ClusterJoiningBody.h"
 #include "qpid/framing/ClusterConnectionDeliverCloseBody.h"
 #include "qpid/framing/ClusterConnectionDeliverDoOutputBody.h"
 #include "qpid/log/Statement.h"
@@ -50,7 +50,13 @@ struct ClusterOperations : public AMQP_AllOperations::ClusterHandler {
     Cluster& cluster;
     MemberId member;
     ClusterOperations(Cluster& c, const MemberId& id) : cluster(c), member(id) {}
-    void urlNotice(const std::string& u) { cluster.urlNotice (member, u); }
+    void joining(const std::string& u) { cluster.joining (member, u); }
+    void ready() { cluster.ready(member); }
+
+    void members(const framing::FieldTable& , const framing::FieldTable& , const framing::FieldTable& ) {
+        assert(0); // Not passed to cluster, used to start a brain dump over TCP.
+    }
+
     bool invoke(AMQFrame& f) { return framing::invoke(*this, *f.getBody()).wasHandled(); }
 };
 
@@ -237,7 +243,7 @@ void Cluster::configChange(
     
     if (nJoined)                // Notfiy new members of my URL.
         mcastFrame(
-            AMQFrame(in_place<ClusterUrlNoticeBody>(ProtocolVersion(), url.str())),
+            AMQFrame(in_place<ClusterJoiningBody>(ProtocolVersion(), url.str())),
             ConnectionId(self,0));
 
 
@@ -261,11 +267,15 @@ void Cluster::disconnect(sys::DispatchHandle& h) {
     broker->shutdown();
 }
 
-void Cluster::urlNotice(const MemberId& m, const string& url) {
+void Cluster::joining(const MemberId& m, const string& url) {
     QPID_LOG(notice, "Cluster member " << m << " has URL " << url);
     urls.insert(UrlMap::value_type(m,Url(url)));
 }
 
+void Cluster::ready(const MemberId& ) {
+    // FIXME aconway 2008-09-08: TODO
+}
+    
 }} // namespace qpid::cluster
 
 
diff --git a/cpp/src/qpid/cluster/Cluster.h b/cpp/src/qpid/cluster/Cluster.h
index 0cd916d5fb..a25b62ea12 100644
--- a/cpp/src/qpid/cluster/Cluster.h
+++ b/cpp/src/qpid/cluster/Cluster.h
@@ -75,7 +75,8 @@ class Cluster : public RefCounted, private Cpg::Handler
     /** Leave the cluster */
     void leave();
     
-    void urlNotice(const MemberId&, const std::string& url);
+    void joining(const MemberId&, const std::string& url);
+    void ready(const MemberId&);
 
     broker::Broker& getBroker() { assert(broker); return *broker; }
 
diff --git a/cpp/src/qpid/cluster/ClusterPlugin.cpp b/cpp/src/qpid/cluster/ClusterPlugin.cpp
index d829683000..31447f2fd0 100644
--- a/cpp/src/qpid/cluster/ClusterPlugin.cpp
+++ b/cpp/src/qpid/cluster/ClusterPlugin.cpp
@@ -26,6 +26,7 @@
 #include "qpid/Plugin.h"
 #include "qpid/Options.h"
 #include "qpid/shared_ptr.h"
+#include "qpid/log/Statement.h"
 
 #include <boost/utility/in_place_factory.hpp>
 
@@ -75,7 +76,7 @@ struct ClusterPlugin : public Plugin {
     void initialize(Plugin::Target& target) {
         broker::Broker* broker = dynamic_cast<broker::Broker*>(&target);
         if (!broker || values.name.empty()) return;  // Only if --cluster-name option was specified.
-        if (cluster) throw Exception("Cluster plugin cannot be initialized twice in one process.");
+        QPID_LOG_IF(warning, cluster, "Ignoring multiple initialization of cluster plugin.");
         cluster = new Cluster(values.name, values.getUrl(broker->getPort()), *broker);
         broker->addFinalizer(boost::bind(&ClusterPlugin::shutdown, this));
         broker->setConnectionFactory(
diff --git a/cpp/src/qpid/cluster/Connection.h b/cpp/src/qpid/cluster/Connection.h
index e6372e80ea..b3e151ce51 100644
--- a/cpp/src/qpid/cluster/Connection.h
+++ b/cpp/src/qpid/cluster/Connection.h
@@ -89,6 +89,18 @@ class Connection :
     // ConnectionInputHandlerFactory
     sys::ConnectionInputHandler* create(sys::ConnectionOutputHandler* out, const std::string& id, bool isClient);
 
+    // State dump methods.
+    virtual void sessionState(const framing::SequenceNumber& /*replayId*/,
+                              const framing::SequenceNumber& /*sendId*/,
+                              const framing::SequenceSet& /*sentIncomplete*/,
+                              const framing::SequenceNumber& /*expectedId*/,
+                              const framing::SequenceNumber& /*receivedId*/,
+                              const framing::SequenceSet& /*unknownCompleted*/,
+                              const framing::SequenceSet& /*receivedIncomplete*/) {}
+    
+    virtual void shadowReady(uint64_t /*clusterId*/,
+                             const std::string& /*userId*/) {}
+
   private:
     void sendDoOutput();
 
diff --git a/cpp/src/qpid/cluster/DumpClient.cpp b/cpp/src/qpid/cluster/DumpClient.cpp
new file mode 100644
index 0000000000..5b92552209
--- /dev/null
+++ b/cpp/src/qpid/cluster/DumpClient.cpp
@@ -0,0 +1,107 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "DumpClient.h"
+#include "qpid/client/SessionBase_0_10Access.h" 
+#include "qpid/broker/Broker.h"
+#include "qpid/broker/Queue.h"
+#include "qpid/broker/QueueRegistry.h"
+#include "qpid/broker/Message.h"
+#include "qpid/broker/Exchange.h"
+#include "qpid/broker/ExchangeRegistry.h"
+#include "qpid/framing/MessageTransferBody.h"
+#include "qpid/framing/enum.h"
+#include "qpid/Url.h"
+#include <boost/bind.hpp>
+
+namespace qpid {
+namespace cluster {
+
+using broker::Broker;
+using broker::Exchange;
+using broker::Queue;
+using broker::QueueBinding;
+using broker::Message;
+using namespace framing::message;
+
+using namespace client;
+
+DumpClient::DumpClient(const Url& url) {
+    connection.open(url);
+    session = connection.newSession();
+}
+
+DumpClient::~DumpClient() {
+    session.close();
+    connection.close();
+}
+
+// Catch-up exchange name: an illegal AMQP exchange name to avoid clashes.
+static const char CATCH_UP_CHARS[] = "\000qpid-dump-exchange";
+static const std::string CATCH_UP(CATCH_UP_CHARS, sizeof(CATCH_UP_CHARS)); 
+
+void DumpClient::dump(Broker& donor) {
+    // TODO aconway 2008-09-08: Caller must handle exceptions
+    // FIXME aconway 2008-09-08: send cluster map frame first.
+    donor.getExchanges().eachExchange(boost::bind(&DumpClient::dumpExchange, this, _1));
+    // Catch-up exchange is used to route messages to the proper queue without modifying routing key.
+    session.exchangeDeclare(arg::exchange=CATCH_UP, arg::type="fanout", arg::autoDelete=true);
+    donor.getQueues().eachQueue(boost::bind(&DumpClient::dumpQueue, this, _1));
+    session.sync();
+}
+
+void DumpClient::dumpExchange(const boost::shared_ptr<Exchange>& ex) {
+    session.exchangeDeclare(
+        ex->getName(), ex->getType(),
+        ex->getAlternate() ? ex->getAlternate()->getName() : std::string(),
+        arg::passive=false,
+        arg::durable=ex->isDurable(),
+        arg::autoDelete=false,
+        arg::arguments=ex->getArgs());
+}
+
+void DumpClient::dumpQueue(const boost::shared_ptr<Queue>& q) {
+    session.queueDeclare(
+        q->getName(),
+        q->getAlternateExchange() ? q->getAlternateExchange()->getName() : std::string(),
+        arg::passive=false,
+        arg::durable=q->isDurable(),
+        arg::exclusive=q->hasExclusiveConsumer(),
+        arg::autoDelete=q->isAutoDelete(),
+        arg::arguments=q->getSettings());
+
+    session.exchangeBind(q->getName(), CATCH_UP, std::string());
+    q->eachMessage(boost::bind(&DumpClient::dumpMessage, this, _1));
+    session.exchangeUnbind(q->getName(), CATCH_UP, std::string());
+    q->eachBinding(boost::bind(&DumpClient::dumpBinding, this, q->getName(), _1));
+}
+
+void DumpClient::dumpMessage(const broker::QueuedMessage& message) {
+    SessionBase_0_10Access sb(session);
+    framing::MessageTransferBody transfer(framing::ProtocolVersion(), CATCH_UP, ACCEPT_MODE_NONE, ACQUIRE_MODE_PRE_ACQUIRED);
+    sb.get()->send(transfer, message.payload->getFrames());
+}
+
+void DumpClient::dumpBinding(const std::string& queue, const QueueBinding& binding) {
+    session.exchangeBind(queue, binding.exchange, binding.key, binding.args);
+}
+
+
+}} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/DumpClient.h b/cpp/src/qpid/cluster/DumpClient.h
new file mode 100644
index 0000000000..447fd1abef
--- /dev/null
+++ b/cpp/src/qpid/cluster/DumpClient.h
@@ -0,0 +1,74 @@
+#ifndef QPID_CLUSTER_DUMPCLIENT_H
+#define QPID_CLUSTER_DUMPCLIENT_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/client/Connection.h"
+#include "qpid/client/AsyncSession.h"
+#include "qpid/broker/Message.h"
+#include "qpid/broker/Queue.h"
+#include "qpid/broker/Exchange.h"
+#include "qpid/broker/QueueRegistry.h"
+#include "qpid/broker/ExchangeRegistry.h"
+#include <boost/shared_ptr.hpp>
+
+
+namespace qpid {
+
+class Url;
+
+namespace broker {
+
+class Broker;
+class Queue;
+class Exchange;
+class QueueBindings;
+class QueueBinding;
+class QueuedMessage;
+} // namespace broker
+
+namespace cluster {
+
+/**
+ * A client that dumps the contents of a local broker to a remote one using AMQP.
+ */
+class DumpClient {
+  public:
+    DumpClient(const Url& receiver);
+    ~DumpClient();
+    
+    void dump(broker::Broker& donor);
+
+  private:
+    void dumpQueue(const boost::shared_ptr<broker::Queue>&);
+    void dumpExchange(const boost::shared_ptr<broker::Exchange>&);
+    void dumpMessage(const broker::QueuedMessage&);
+    void dumpBinding(const std::string& queue, const broker::QueueBinding& binding);
+
+  private:
+    client::Connection connection;
+    client::AsyncSession session;
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_DUMPCLIENT_H*/