QPID-1567: Initial support for asynchronous queue state replication

* Added QueueEvents class with per broker instance
* Modified qpid::broker::Queue to notify QueueEvents of enqueues and dequeues (based on configuration)
* Added replication subdir containing two plugins:
  - an event listener that registers with QueueEvents and creates messages representing received 
    events on a replication queue
  - a custom exchange type for processing messages of the format created by the listener plugin
* Added new option for controlling event generation to qpid::client::QueueOptions
* Added new queue option to qpid-config script for the same



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

5 files changed, 417 insertions, 0 deletions

diff --git a/cpp/src/qpid/replication/ReplicatingEventListener.cpp b/cpp/src/qpid/replication/ReplicatingEventListener.cpp
new file mode 100644
index 0000000000..80ff77d107
--- /dev/null
+++ b/cpp/src/qpid/replication/ReplicatingEventListener.cpp
@@ -0,0 +1,122 @@
+/*
+ *
+ * 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 "ReplicatingEventListener.h"
+#include "constants.h"
+#include "qpid/broker/Broker.h"
+#include "qpid/broker/QueueEvents.h"
+#include "qpid/framing/AMQFrame.h"
+#include "qpid/framing/MessageTransferBody.h"
+#include "qpid/log/Statement.h"
+
+namespace qpid {
+namespace replication {
+
+using namespace qpid::broker;
+using namespace qpid::framing;
+using namespace qpid::replication::constants;
+
+void ReplicatingEventListener::handle(QueueEvents::Event event)
+{
+    //create event message and enqueue it on replication queue
+    FieldTable headers;
+    boost::intrusive_ptr<Message> message;
+    switch (event.type) {
+      case QueueEvents::ENQUEUE:
+        headers.setString(REPLICATION_EVENT_TYPE, ENQUEUE);
+        headers.setString(REPLICATION_TARGET_QUEUE, event.msg.queue->getName());
+        message = createEventMessage(headers);
+        queue->deliver(message);
+        //if its an enqueue, enqueue the message itself on the
+        //replication queue also:
+        queue->deliver(event.msg.payload);
+        QPID_LOG(debug, "Queued 'enqueue' event on " << event.msg.queue->getName() << " for replication");
+        break;
+      case QueueEvents::DEQUEUE:
+        headers.setString(REPLICATION_EVENT_TYPE, DEQUEUE);
+        headers.setString(REPLICATION_TARGET_QUEUE, event.msg.queue->getName());
+        headers.setInt(DEQUEUED_MESSAGE_POSITION, event.msg.position);
+        message = createEventMessage(headers);
+        queue->deliver(message);
+        QPID_LOG(debug, "Queued 'dequeue' event from " << event.msg.queue->getName() << " for replication, (from position "
+                 << event.msg.position << ")");
+        break;
+    }
+}
+
+namespace {
+const std::string EMPTY;
+}
+
+boost::intrusive_ptr<Message> ReplicatingEventListener::createEventMessage(const FieldTable& headers)
+{
+        boost::intrusive_ptr<Message> msg(new Message());
+        AMQFrame method(in_place<MessageTransferBody>(ProtocolVersion(), EMPTY, 0, 0));
+        AMQFrame header(in_place<AMQHeaderBody>());
+        header.setBof(false);
+        header.setEof(true);
+        header.setBos(true);
+        header.setEos(true);
+        msg->getFrames().append(method);
+        msg->getFrames().append(header);
+        MessageProperties* props = msg->getFrames().getHeaders()->get<MessageProperties>(true);
+        props->setApplicationHeaders(headers);
+        return msg;
+}
+    
+Options* ReplicatingEventListener::getOptions() 
+{ 
+    return &options; 
+}
+
+void ReplicatingEventListener::initialize(Plugin::Target& target)
+{
+      Broker* broker = dynamic_cast<broker::Broker*>(&target);
+      if (broker && !options.queue.empty()) {
+          if (options.createQueue) {
+              queue = broker->getQueues().declare(options.queue).first;
+          } else {
+              queue = broker->getQueues().find(options.queue);
+          }
+          if (queue) {
+              QueueEvents::EventListener callback = boost::bind(&ReplicatingEventListener::handle, this, _1);
+              broker->getQueueEvents().registerListener(options.name, callback);
+              QPID_LOG(info, "Registered replicating queue event listener");
+          } else {
+              QPID_LOG(error, "Replication queue named '" << options.queue << "' does not exist; replication plugin disabled.");
+          }
+      }
+}
+
+void ReplicatingEventListener::earlyInitialize(Target&) {}
+
+ReplicatingEventListener::PluginOptions::PluginOptions() : Options("Queue Replication Options"), 
+                                                           name("replicator"), 
+                                                           createQueue(false)
+{
+    addOptions()
+        ("replication-queue", optValue(queue, "QUEUE"), "Queue on which events for other queues are recorded")
+        ("replication-listener-name", optValue(name, "NAME"), "name by which to register the replicating event listener")
+        ("create-replication-queue", optValue(createQueue), "if set, the replication will be created if it does not exist");
+}
+
+static ReplicatingEventListener plugin;
+
+}} // namespace qpid::replication
diff --git a/cpp/src/qpid/replication/ReplicatingEventListener.h b/cpp/src/qpid/replication/ReplicatingEventListener.h
new file mode 100644
index 0000000000..25e2a5b7b9
--- /dev/null
+++ b/cpp/src/qpid/replication/ReplicatingEventListener.h
@@ -0,0 +1,66 @@
+#ifndef QPID_REPLICATION_REPLICATINGEVENTLISTENER_H
+#define QPID_REPLICATION_REPLICATINGEVENTLISTENER_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/Plugin.h"
+#include "qpid/Options.h"
+#include "qpid/broker/Message.h"
+#include "qpid/broker/Queue.h"
+#include "qpid/broker/QueueEvents.h"
+#include "qpid/framing/FieldTable.h"
+
+namespace qpid {
+namespace replication {
+
+/**
+ * An event listener plugin that records queue events as messages on a
+ * replication queue, from where they can be consumed (e.g. by an
+ * inter-broker link to the corresponding QueueReplicationExchange
+ * plugin.
+ */
+class ReplicatingEventListener : public Plugin
+{
+  public:
+    Options* getOptions();
+    void earlyInitialize(Plugin::Target& target);
+    void initialize(Plugin::Target& target);
+    void handle(qpid::broker::QueueEvents::Event);
+  private:    
+    struct PluginOptions : public Options
+    {
+        std::string queue;
+        std::string name;
+        bool createQueue;
+
+        PluginOptions();
+    };
+
+    PluginOptions options;    
+    qpid::broker::Queue::shared_ptr queue;
+
+    boost::intrusive_ptr<qpid::broker::Message> createEventMessage(const qpid::framing::FieldTable& headers);
+};
+
+}} // namespace qpid::replication
+
+#endif  /*!QPID_REPLICATION_REPLICATINGEVENTLISTENER_H*/
diff --git a/cpp/src/qpid/replication/ReplicationExchange.cpp b/cpp/src/qpid/replication/ReplicationExchange.cpp
new file mode 100644
index 0000000000..abe8a4dfb6
--- /dev/null
+++ b/cpp/src/qpid/replication/ReplicationExchange.cpp
@@ -0,0 +1,139 @@
+/*
+ *
+ * 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 "ReplicationExchange.h"
+#include "constants.h"
+#include "qpid/Plugin.h"
+#include "qpid/broker/Broker.h"
+#include "qpid/broker/ExchangeRegistry.h"
+#include "qpid/framing/reply_exceptions.h"
+#include "qpid/log/Statement.h"
+#include <boost/bind.hpp>
+
+namespace qpid {
+namespace replication {
+
+using namespace qpid::broker;
+using namespace qpid::framing;
+using namespace qpid::replication::constants;
+
+ReplicationExchange::ReplicationExchange(const std::string& name, bool durable, 
+                                         const FieldTable& args,
+                                         QueueRegistry& qr,
+                                         Manageable* parent) 
+    : Exchange(name, durable, args, parent), queues(qr), expectingEnqueue(false) {}
+
+std::string ReplicationExchange::getType() const { return typeName; }            
+
+void ReplicationExchange::route(Deliverable& msg, const std::string& /*routingKey*/, const FieldTable* args)
+{
+    if (args) {
+        std::string eventType = args->getAsString(REPLICATION_EVENT_TYPE);
+        if (eventType == ENQUEUE) {
+            expectingEnqueue = true;
+            targetQueue = args->getAsString(REPLICATION_TARGET_QUEUE);
+            QPID_LOG(debug, "Recorded replicated 'enqueue' event for " << targetQueue);
+            return;
+        } else if (eventType == DEQUEUE) {
+            std::string queueName = args->getAsString(REPLICATION_TARGET_QUEUE);
+            Queue::shared_ptr queue = queues.find(queueName);
+            SequenceNumber position(args->getAsInt(DEQUEUED_MESSAGE_POSITION));
+
+            QueuedMessage dequeued;
+            if (queue->acquireMessageAt(position, dequeued)) {
+                queue->dequeue(0, dequeued);
+                QPID_LOG(debug, "Processed replicated 'dequeue' event from " << queueName << " at position " << position);
+            } else {
+                QPID_LOG(warning, "Could not acquire message " << position << " from " << queueName);
+            }
+            
+            return;
+        } else if (!eventType.empty()) {
+            throw IllegalArgumentException(QPID_MSG("Illegal value for " << REPLICATION_EVENT_TYPE << ": " << eventType));
+        }
+    }
+    //if we get here assume its not an event message, assume its an enqueue
+    if (expectingEnqueue) {
+        Queue::shared_ptr queue = queues.find(targetQueue);
+        msg.deliverTo(queue);
+        expectingEnqueue = false;
+        targetQueue.clear();
+        QPID_LOG(debug, "Eenqueued replicated message onto " << targetQueue);
+    } else {
+        QPID_LOG(warning, "Dropping unexpected message");
+    }
+}
+
+bool ReplicationExchange::bind(Queue::shared_ptr /*queue*/, const std::string& /*routingKey*/, const FieldTable* /*args*/)
+{
+    throw NotImplementedException("Replication exchange does not support bind operation");
+}
+
+bool ReplicationExchange::unbind(Queue::shared_ptr /*queue*/, const std::string& /*routingKey*/, const FieldTable* /*args*/)
+{
+    throw NotImplementedException("Replication exchange does not support unbind operation");
+}
+
+bool ReplicationExchange::isBound(Queue::shared_ptr /*queue*/, const string* const /*routingKey*/, const FieldTable* const /*args*/)
+{
+    return false;
+}
+
+const std::string ReplicationExchange::typeName("replication");
+
+
+struct ReplicationExchangePlugin : Plugin
+{
+    Broker* broker;
+
+    ReplicationExchangePlugin();
+    void earlyInitialize(Plugin::Target& target);
+    void initialize(Plugin::Target& target);
+    Exchange::shared_ptr create(const std::string& name, bool durable,
+                                const framing::FieldTable& args, 
+                                management::Manageable* parent);
+};
+
+ReplicationExchangePlugin::ReplicationExchangePlugin() : broker(0) {}
+
+Exchange::shared_ptr ReplicationExchangePlugin::create(const std::string& name, bool durable,
+                                                       const framing::FieldTable& args, 
+                                                       management::Manageable* parent)
+{
+    Exchange::shared_ptr e(new ReplicationExchange(name, durable, args, broker->getQueues(), parent));
+    return e;
+}
+
+
+void ReplicationExchangePlugin::initialize(Plugin::Target& target)
+{
+    broker = dynamic_cast<broker::Broker*>(&target);
+    if (broker) {
+        ExchangeRegistry::FactoryFunction f = boost::bind(&ReplicationExchangePlugin::create, this, _1, _2, _3, _4);
+        broker->getExchanges().registerType(ReplicationExchange::typeName, f);
+        QPID_LOG(info, "Registered replication exchange");
+    }
+}
+
+void ReplicationExchangePlugin::earlyInitialize(Target&) {}
+
+static ReplicationExchangePlugin exchangePlugin;
+
+}} // namespace qpid::replication
diff --git a/cpp/src/qpid/replication/ReplicationExchange.h b/cpp/src/qpid/replication/ReplicationExchange.h
new file mode 100644
index 0000000000..ed2b5956b6
--- /dev/null
+++ b/cpp/src/qpid/replication/ReplicationExchange.h
@@ -0,0 +1,59 @@
+#ifndef QPID_REPLICATION_REPLICATIONEXCHANGE_H
+#define QPID_REPLICATION_REPLICATIONEXCHANGE_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/broker/Exchange.h"
+
+namespace qpid {
+namespace replication {
+
+/**
+ * A custom exchange plugin that processes incoming messages
+ * representing enqueue or dequeue events for particular queues and
+ * carries out the corresponding action to replicate that on the local
+ * broker.
+ */
+class ReplicationExchange : public qpid::broker::Exchange
+{
+  public:
+    static const std::string typeName;
+
+    ReplicationExchange(const std::string& name, bool durable, 
+                        const qpid::framing::FieldTable& args,
+                        qpid::broker::QueueRegistry& queues,
+                        qpid::management::Manageable* parent = 0);
+
+    std::string getType() const;
+
+    void route(qpid::broker::Deliverable& msg, const std::string& routingKey, const qpid::framing::FieldTable* args);
+        
+    bool bind(qpid::broker::Queue::shared_ptr queue, const std::string& routingKey, const qpid::framing::FieldTable* args);
+    bool unbind(qpid::broker::Queue::shared_ptr queue, const std::string& routingKey, const qpid::framing::FieldTable* args);
+    bool isBound(qpid::broker::Queue::shared_ptr queue, const std::string* const routingKey, const qpid::framing::FieldTable* const args);
+  private:
+    qpid::broker::QueueRegistry& queues;
+    bool expectingEnqueue;
+    std::string targetQueue;
+};
+}} // namespace qpid::replication
+
+#endif  /*!QPID_REPLICATION_REPLICATIONEXCHANGE_H*/
diff --git a/cpp/src/qpid/replication/constants.h b/cpp/src/qpid/replication/constants.h
new file mode 100644
index 0000000000..b0cef7570c
--- /dev/null
+++ b/cpp/src/qpid/replication/constants.h
@@ -0,0 +1,31 @@
+/*
+ *
+ * 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.
+ *
+ */
+namespace qpid {
+namespace replication {
+namespace constants {
+
+const std::string REPLICATION_EVENT_TYPE("qpid.replication_event_type");
+const std::string ENQUEUE("enqueue");
+const std::string DEQUEUE("dequeue");
+const std::string REPLICATION_TARGET_QUEUE("qpid.replication_target_queue");
+const std::string DEQUEUED_MESSAGE_POSITION("qpid.dequeued_message_position");
+
+}}}