diff options
Diffstat (limited to 'qpid/cpp/src/qpid/replication')
4 files changed, 36 insertions, 17 deletions
diff --git a/qpid/cpp/src/qpid/replication/ReplicatingEventListener.cpp b/qpid/cpp/src/qpid/replication/ReplicatingEventListener.cpp index 2d8af3b052..e3990a13cc 100644 --- a/qpid/cpp/src/qpid/replication/ReplicatingEventListener.cpp +++ b/qpid/cpp/src/qpid/replication/ReplicatingEventListener.cpp @@ -57,7 +57,6 @@ void ReplicatingEventListener::deliverDequeueMessage(const QueuedMessage& dequeu { FieldTable headers; headers.setString(REPLICATION_TARGET_QUEUE, dequeued.queue->getName()); - headers.setInt(REPLICATION_EVENT_SEQNO, ++sequence); headers.setInt(REPLICATION_EVENT_TYPE, DEQUEUE); headers.setInt(DEQUEUED_MESSAGE_POSITION, dequeued.position); boost::intrusive_ptr<Message> msg(createMessage(headers)); @@ -69,7 +68,6 @@ void ReplicatingEventListener::deliverEnqueueMessage(const QueuedMessage& enqueu boost::intrusive_ptr<Message> msg(cloneMessage(*(enqueued.queue), enqueued.payload)); FieldTable& headers = msg->getProperties<MessageProperties>()->getApplicationHeaders(); headers.setString(REPLICATION_TARGET_QUEUE, enqueued.queue->getName()); - headers.setInt(REPLICATION_EVENT_SEQNO, ++sequence); headers.setInt(REPLICATION_EVENT_TYPE, ENQUEUE); queue->deliver(msg); } @@ -131,12 +129,14 @@ void ReplicatingEventListener::initialize(Plugin::Target& target) { Broker* broker = dynamic_cast<broker::Broker*>(&target); if (broker && !options.queue.empty()) { + broker->addFinalizer(boost::bind(&ReplicatingEventListener::shutdown, this)); if (options.createQueue) { queue = broker->getQueues().declare(options.queue).first; } else { queue = broker->getQueues().find(options.queue); } if (queue) { + queue->insertSequenceNumbers(REPLICATION_EVENT_SEQNO); QueueEvents::EventListener callback = boost::bind(&ReplicatingEventListener::handle, this, _1); broker->getQueueEvents().registerListener(options.name, callback); QPID_LOG(info, "Registered replicating queue event listener"); @@ -147,6 +147,7 @@ void ReplicatingEventListener::initialize(Plugin::Target& target) } void ReplicatingEventListener::earlyInitialize(Target&) {} +void ReplicatingEventListener::shutdown() { queue.reset(); } ReplicatingEventListener::PluginOptions::PluginOptions() : Options("Queue Replication Options"), name("replicator"), diff --git a/qpid/cpp/src/qpid/replication/ReplicatingEventListener.h b/qpid/cpp/src/qpid/replication/ReplicatingEventListener.h index 7616c7ac8a..3d8f23e7ac 100644 --- a/qpid/cpp/src/qpid/replication/ReplicatingEventListener.h +++ b/qpid/cpp/src/qpid/replication/ReplicatingEventListener.h @@ -58,10 +58,10 @@ class ReplicatingEventListener : public Plugin PluginOptions options; qpid::broker::Queue::shared_ptr queue; - qpid::framing::SequenceNumber sequence; void deliverDequeueMessage(const qpid::broker::QueuedMessage& enqueued); void deliverEnqueueMessage(const qpid::broker::QueuedMessage& enqueued); + void shutdown(); boost::intrusive_ptr<qpid::broker::Message> createMessage(const qpid::framing::FieldTable& headers); boost::intrusive_ptr<qpid::broker::Message> cloneMessage(qpid::broker::Queue& queue, diff --git a/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp b/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp index 639cfb5d2e..88c94ad7ba 100644 --- a/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp +++ b/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp @@ -34,11 +34,13 @@ using namespace qpid::broker; using namespace qpid::framing; using namespace qpid::replication::constants; +const std::string SEQUENCE_VALUE("qpid.replication-event.sequence"); ReplicationExchange::ReplicationExchange(const std::string& name, bool durable, const FieldTable& args, QueueRegistry& qr, Manageable* parent) - : Exchange(name, durable, args, parent), queues(qr), init(false) {} + : Exchange(name, durable, args, parent), queues(qr), sequence(args.getAsInt64(SEQUENCE_VALUE)), init(false) + {} std::string ReplicationExchange::getType() const { return typeName; } @@ -68,26 +70,33 @@ void ReplicationExchange::handleEnqueueEvent(const FieldTable* args, Deliverable { std::string queueName = args->getAsString(REPLICATION_TARGET_QUEUE); Queue::shared_ptr queue = queues.find(queueName); - FieldTable& headers = msg.getMessage().getProperties<MessageProperties>()->getApplicationHeaders(); - headers.erase(REPLICATION_TARGET_QUEUE); - headers.erase(REPLICATION_EVENT_SEQNO); - headers.erase(REPLICATION_EVENT_TYPE); - msg.deliverTo(queue); - QPID_LOG(debug, "Enqueued replicated message onto " << queue); + if (queue) { + FieldTable& headers = msg.getMessage().getProperties<MessageProperties>()->getApplicationHeaders(); + headers.erase(REPLICATION_TARGET_QUEUE); + headers.erase(REPLICATION_EVENT_SEQNO); + headers.erase(REPLICATION_EVENT_TYPE); + msg.deliverTo(queue); + QPID_LOG(debug, "Enqueued replicated message onto " << queueName); + } else { + QPID_LOG(error, "Cannot enqueue replicated message. Queue " << queueName << " does not exist"); + } } void ReplicationExchange::handleDequeueEvent(const FieldTable* args) { 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); + if (queue) { + 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); + } } else { - QPID_LOG(warning, "Could not acquire message " << position << " from " << queueName); + QPID_LOG(error, "Cannot process replicated 'dequeue' event. Queue " << queueName << " does not exist"); } } @@ -128,6 +137,13 @@ bool ReplicationExchange::isBound(Queue::shared_ptr /*queue*/, const string* con const std::string ReplicationExchange::typeName("replication"); +void ReplicationExchange::encode(Buffer& buffer) const +{ + args.setInt64(std::string(SEQUENCE_VALUE), sequence); + Exchange::encode(buffer); +} + + struct ReplicationExchangePlugin : Plugin { Broker* broker; diff --git a/qpid/cpp/src/qpid/replication/ReplicationExchange.h b/qpid/cpp/src/qpid/replication/ReplicationExchange.h index 897e4a954e..4cc45ed5f5 100644 --- a/qpid/cpp/src/qpid/replication/ReplicationExchange.h +++ b/qpid/cpp/src/qpid/replication/ReplicationExchange.h @@ -22,6 +22,7 @@ * */ #include "qpid/broker/Exchange.h" +#include "qpid/framing/Buffer.h" #include "qpid/framing/SequenceNumber.h" namespace qpid { @@ -58,6 +59,7 @@ class ReplicationExchange : public qpid::broker::Exchange bool isDuplicate(const qpid::framing::FieldTable* args); void handleEnqueueEvent(const qpid::framing::FieldTable* args, qpid::broker::Deliverable& msg); void handleDequeueEvent(const qpid::framing::FieldTable* args); + void encode(framing::Buffer& buffer) const; }; }} // namespace qpid::replication |