summaryrefslogtreecommitdiff
path: root/cpp/src/qpid
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/qpid')
-rw-r--r--cpp/src/qpid/broker/Queue.cpp8
-rw-r--r--cpp/src/qpid/replication/ReplicatingEventListener.cpp78
-rw-r--r--cpp/src/qpid/replication/ReplicatingEventListener.h5
-rw-r--r--cpp/src/qpid/replication/ReplicationExchange.cpp1
4 files changed, 71 insertions, 21 deletions
diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index a1a83926bf..aa0cd8ca31 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -32,6 +32,7 @@
#include "qpid/log/Statement.h"
#include "qpid/management/ManagementBroker.h"
#include "qpid/framing/reply_exceptions.h"
+#include "qpid/framing/FieldTable.h"
#include "qpid/sys/Monitor.h"
#include "qpid/sys/Time.h"
#include "qmf/org/apache/qpid/broker/ArgsQueuePurge.h"
@@ -68,6 +69,9 @@ const std::string qpidLastValueQueueNoBrowse("qpid.last_value_queue_no_browse");
const std::string qpidPersistLastNode("qpid.persist_last_node");
const std::string qpidVQMatchProperty("qpid.LVQ_key");
const std::string qpidQueueEventGeneration("qpid.queue_event_generation");
+//following feature is not ready for general use as it doesn't handle
+//the case where a message is enqueued on more than one queue well enough:
+const std::string qpidInsertSequenceNumbers("qpid.insert_sequence_numbers");
const int ENQUEUE_ONLY=1;
const int ENQUEUE_AND_DEQUEUE=2;
@@ -774,6 +778,9 @@ void Queue::configure(const FieldTable& _settings, bool recovering)
eventMode = _settings.getAsInt(qpidQueueEventGeneration);
+ FieldTable::ValuePtr p =_settings.get(qpidInsertSequenceNumbers);
+ if (p && p->convertsTo<std::string>()) insertSequenceNumbers(p->get<std::string>());
+
if (mgmtObject != 0)
mgmtObject->set_arguments (_settings);
@@ -997,4 +1004,5 @@ void Queue::insertSequenceNumbers(const std::string& key)
{
seqNoKey = key;
insertSeqNo = !seqNoKey.empty();
+ QPID_LOG(debug, "Inserting sequence numbers as " << key);
}
diff --git a/cpp/src/qpid/replication/ReplicatingEventListener.cpp b/cpp/src/qpid/replication/ReplicatingEventListener.cpp
index e3990a13cc..60d5fffc0b 100644
--- a/cpp/src/qpid/replication/ReplicatingEventListener.cpp
+++ b/cpp/src/qpid/replication/ReplicatingEventListener.cpp
@@ -21,6 +21,7 @@
#include "ReplicatingEventListener.h"
#include "constants.h"
#include "qpid/broker/Broker.h"
+#include "qpid/broker/DeliverableMessage.h"
#include "qpid/broker/QueueEvents.h"
#include "qpid/framing/AMQFrame.h"
#include "qpid/framing/FrameHandler.h"
@@ -60,7 +61,9 @@ void ReplicatingEventListener::deliverDequeueMessage(const QueuedMessage& dequeu
headers.setInt(REPLICATION_EVENT_TYPE, DEQUEUE);
headers.setInt(DEQUEUED_MESSAGE_POSITION, dequeued.position);
boost::intrusive_ptr<Message> msg(createMessage(headers));
- queue->deliver(msg);
+ DeliveryProperties* props = msg->getFrames().getHeaders()->get<DeliveryProperties>(true);
+ props->setRoutingKey(dequeued.queue->getName());
+ route(msg);
}
void ReplicatingEventListener::deliverEnqueueMessage(const QueuedMessage& enqueued)
@@ -69,9 +72,26 @@ void ReplicatingEventListener::deliverEnqueueMessage(const QueuedMessage& enqueu
FieldTable& headers = msg->getProperties<MessageProperties>()->getApplicationHeaders();
headers.setString(REPLICATION_TARGET_QUEUE, enqueued.queue->getName());
headers.setInt(REPLICATION_EVENT_TYPE, ENQUEUE);
- queue->deliver(msg);
+ route(msg);
}
+void ReplicatingEventListener::route(boost::intrusive_ptr<qpid::broker::Message> msg)
+{
+ try {
+ if (exchange) {
+ DeliverableMessage deliverable(msg);
+ exchange->route(deliverable, msg->getRoutingKey(), msg->getApplicationHeaders());
+ } else if (queue) {
+ queue->deliver(msg);
+ } else {
+ QPID_LOG(error, "Cannot route replication event, neither replication queue nor exchange configured");
+ }
+ } catch (const std::exception& e) {
+ QPID_LOG(error, "Error enqueing replication event: " << e.what());
+ }
+}
+
+
boost::intrusive_ptr<Message> ReplicatingEventListener::createMessage(const FieldTable& headers)
{
boost::intrusive_ptr<Message> msg(new Message());
@@ -127,33 +147,49 @@ Options* ReplicatingEventListener::getOptions()
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");
- } else {
- QPID_LOG(error, "Replication queue named '" << options.queue << "' does not exist; replication plugin disabled.");
- }
- }
+ Broker* broker = dynamic_cast<broker::Broker*>(&target);
+ if (broker) {
+ broker->addFinalizer(boost::bind(&ReplicatingEventListener::shutdown, this));
+ if (!options.exchange.empty()) {
+ if (!options.queue.empty()) {
+ QPID_LOG(warning, "Replication queue option ignored as replication exchange has been specified");
+ }
+ try {
+ exchange = broker->getExchanges().declare(options.exchange, options.exchangeType).first;
+ } catch (const UnknownExchangeTypeException&) {
+ QPID_LOG(error, "Replication disabled due to invalid type: " << options.exchangeType);
+ }
+ } else if (!options.queue.empty()) {
+ if (options.createQueue) {
+ queue = broker->getQueues().declare(options.queue).first;
+ } else {
+ queue = broker->getQueues().find(options.queue);
+ }
+ if (queue) {
+ queue->insertSequenceNumbers(REPLICATION_EVENT_SEQNO);
+ } else {
+ QPID_LOG(error, "Replication queue named '" << options.queue << "' does not exist; replication plugin disabled.");
+ }
+ }
+ if (queue || exchange) {
+ QueueEvents::EventListener callback = boost::bind(&ReplicatingEventListener::handle, this, _1);
+ broker->getQueueEvents().registerListener(options.name, callback);
+ QPID_LOG(info, "Registered replicating queue event listener");
+ }
+ }
}
void ReplicatingEventListener::earlyInitialize(Target&) {}
-void ReplicatingEventListener::shutdown() { queue.reset(); }
+void ReplicatingEventListener::shutdown() { queue.reset(); exchange.reset(); }
ReplicatingEventListener::PluginOptions::PluginOptions() : Options("Queue Replication Options"),
- name("replicator"),
+ exchangeType("direct"),
+ name("replicator"),
createQueue(false)
{
addOptions()
+ ("replication-exchange-name", optValue(exchange, "EXCHANGE"), "Exchange to which events for other queues are routed")
+ ("replication-exchange-type", optValue(exchangeType, "direct|topic etc"), "Type of exchange to use")
("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");
diff --git a/cpp/src/qpid/replication/ReplicatingEventListener.h b/cpp/src/qpid/replication/ReplicatingEventListener.h
index 3d8f23e7ac..74418d00e6 100644
--- a/cpp/src/qpid/replication/ReplicatingEventListener.h
+++ b/cpp/src/qpid/replication/ReplicatingEventListener.h
@@ -24,6 +24,7 @@
#include "qpid/Plugin.h"
#include "qpid/Options.h"
+#include "qpid/broker/Exchange.h"
#include "qpid/broker/Message.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/QueueEvents.h"
@@ -50,6 +51,8 @@ class ReplicatingEventListener : public Plugin
struct PluginOptions : public Options
{
std::string queue;
+ std::string exchange;
+ std::string exchangeType;
std::string name;
bool createQueue;
@@ -58,9 +61,11 @@ class ReplicatingEventListener : public Plugin
PluginOptions options;
qpid::broker::Queue::shared_ptr queue;
+ qpid::broker::Exchange::shared_ptr exchange;
void deliverDequeueMessage(const qpid::broker::QueuedMessage& enqueued);
void deliverEnqueueMessage(const qpid::broker::QueuedMessage& enqueued);
+ void route(boost::intrusive_ptr<qpid::broker::Message>);
void shutdown();
boost::intrusive_ptr<qpid::broker::Message> createMessage(const qpid::framing::FieldTable& headers);
diff --git a/cpp/src/qpid/replication/ReplicationExchange.cpp b/cpp/src/qpid/replication/ReplicationExchange.cpp
index 88c94ad7ba..29cdf21bc6 100644
--- a/cpp/src/qpid/replication/ReplicationExchange.cpp
+++ b/cpp/src/qpid/replication/ReplicationExchange.cpp
@@ -102,6 +102,7 @@ void ReplicationExchange::handleDequeueEvent(const FieldTable* args)
bool ReplicationExchange::isDuplicate(const FieldTable* args)
{
+ if (!args->get(REPLICATION_EVENT_SEQNO)) return false;
SequenceNumber seqno(args->getAsInt(REPLICATION_EVENT_SEQNO));
if (!init) {
init = true;
View Lorry Controller Status