diff options
Diffstat (limited to 'cpp')
-rw-r--r-- | cpp/src/qpid/broker/Queue.cpp | 8 | ||||
-rw-r--r-- | cpp/src/qpid/replication/ReplicatingEventListener.cpp | 78 | ||||
-rw-r--r-- | cpp/src/qpid/replication/ReplicatingEventListener.h | 5 | ||||
-rw-r--r-- | cpp/src/qpid/replication/ReplicationExchange.cpp | 1 | ||||
-rw-r--r-- | cpp/src/tests/Makefile.am | 3 | ||||
-rw-r--r-- | cpp/src/tests/ReplicationTest.cpp | 129 |
6 files changed, 202 insertions, 22 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; diff --git a/cpp/src/tests/Makefile.am b/cpp/src/tests/Makefile.am index 7a33a7257b..9a81ef18b3 100644 --- a/cpp/src/tests/Makefile.am +++ b/cpp/src/tests/Makefile.am @@ -95,7 +95,8 @@ unit_test_SOURCES= unit_test.cpp unit_test.h \ ProxyTest.cpp \ RetryList.cpp \ RateFlowcontrolTest.cpp \ - FrameDecoder.cpp + FrameDecoder.cpp \ + ReplicationTest.cpp if HAVE_XML unit_test_SOURCES+= XmlClientSessionTest.cpp diff --git a/cpp/src/tests/ReplicationTest.cpp b/cpp/src/tests/ReplicationTest.cpp new file mode 100644 index 0000000000..7bd585639d --- /dev/null +++ b/cpp/src/tests/ReplicationTest.cpp @@ -0,0 +1,129 @@ + /* + * + * 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 "unit_test.h" +#include "test_tools.h" +#include "BrokerFixture.h" + +#include "qpid/Plugin.h" +#include "qpid/broker/Broker.h" +#include "qpid/client/QueueOptions.h" +#include "qpid/framing/reply_exceptions.h" +#include "qpid/framing/SequenceNumber.h" +#include "qpid/replication/constants.h" +#include "qpid/sys/Shlib.h" + +#include <string> +#include <sstream> +#include <vector> + +#include <boost/assign.hpp> +#include <boost/bind.hpp> + +using namespace qpid::client; +using namespace qpid::framing; +using namespace qpid::replication::constants; +using boost::assign::list_of; + +QPID_AUTO_TEST_SUITE(ReplicationTestSuite) + +qpid::sys::Shlib plugin("../.libs/replicating_listener.so"); + +qpid::broker::Broker::Options getBrokerOpts(const std::vector<std::string>& args) +{ + std::vector<const char*> argv(args.size()); + transform(args.begin(), args.end(), argv.begin(), boost::bind(&string::c_str, _1)); + + qpid::broker::Broker::Options opts; + qpid::Plugin::addOptions(opts); + opts.parse(argv.size(), &argv[0], "", true); + return opts; +} + +QPID_AUTO_TEST_CASE(testReplicationExchange) +{ + qpid::broker::Broker::Options brokerOpts(getBrokerOpts(list_of<string>("qpidd") + ("--replication-exchange-name=qpid.replication"))); + ProxySessionFixture f(brokerOpts); + + + std::string dataQ("queue-1"); + std::string eventQ("event-queue-1"); + std::string dataQ2("queue-2"); + std::string eventQ2("event-queue-2"); + FieldTable eventQopts; + eventQopts.setString("qpid.insert_sequence_numbers", REPLICATION_EVENT_SEQNO); + + f.session.queueDeclare(arg::queue=eventQ, arg::exclusive=true, arg::autoDelete=true, arg::arguments=eventQopts); + f.session.exchangeBind(arg::exchange="qpid.replication", arg::queue=eventQ, arg::bindingKey=dataQ); + + f.session.queueDeclare(arg::queue=eventQ2, arg::exclusive=true, arg::autoDelete=true, arg::arguments=eventQopts); + f.session.exchangeBind(arg::exchange="qpid.replication", arg::queue=eventQ2, arg::bindingKey=dataQ2); + + QueueOptions args; + args.enableQueueEvents(false); + f.session.queueDeclare(arg::queue=dataQ, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args); + f.session.queueDeclare(arg::queue=dataQ2, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args); + for (int i = 0; i < 10; i++) { + f.session.messageTransfer(arg::content=Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), dataQ)); + f.session.messageTransfer(arg::content=Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), dataQ2)); + } + Message msg; + LocalQueue incoming; + Subscription sub = f.subs.subscribe(incoming, dataQ); + for (int i = 0; i < 10; i++) { + BOOST_CHECK(incoming.get(msg, qpid::sys::TIME_SEC)); + BOOST_CHECK_EQUAL((boost::format("%1%_%2%") % "Message" % (i+1)).str(), msg.getData()); + } + BOOST_CHECK(!f.subs.get(msg, dataQ)); + + sub.cancel(); + sub = f.subs.subscribe(incoming, eventQ); + //check that we received enqueue events for first queue: + for (int i = 0; i < 10; i++) { + BOOST_CHECK(incoming.get(msg, qpid::sys::TIME_SEC)); + BOOST_CHECK_EQUAL(msg.getHeaders().getAsString(REPLICATION_TARGET_QUEUE), dataQ); + BOOST_CHECK_EQUAL(msg.getHeaders().getAsInt(REPLICATION_EVENT_TYPE), ENQUEUE); + BOOST_CHECK_EQUAL(msg.getHeaders().getAsInt64(REPLICATION_EVENT_SEQNO), (i+1)); + BOOST_CHECK_EQUAL((boost::format("%1%_%2%") % "Message" % (i+1)).str(), msg.getData()); + } + //check that we received dequeue events for first queue: + for (int i = 0; i < 10; i++) { + BOOST_CHECK(incoming.get(msg, qpid::sys::TIME_SEC)); + BOOST_CHECK_EQUAL(msg.getHeaders().getAsString(REPLICATION_TARGET_QUEUE), dataQ); + BOOST_CHECK_EQUAL(msg.getHeaders().getAsInt(REPLICATION_EVENT_TYPE), DEQUEUE); + BOOST_CHECK_EQUAL(msg.getHeaders().getAsInt(DEQUEUED_MESSAGE_POSITION), (i+1)); + BOOST_CHECK_EQUAL(msg.getHeaders().getAsInt64(REPLICATION_EVENT_SEQNO), (i+11)); + } + + sub.cancel(); + sub = f.subs.subscribe(incoming, eventQ2); + //check that we received enqueue events for second queue: + for (int i = 0; i < 10; i++) { + BOOST_CHECK(incoming.get(msg, qpid::sys::TIME_SEC)); + BOOST_CHECK_EQUAL(msg.getHeaders().getAsString(REPLICATION_TARGET_QUEUE), dataQ2); + BOOST_CHECK_EQUAL(msg.getHeaders().getAsInt(REPLICATION_EVENT_TYPE), ENQUEUE); + BOOST_CHECK_EQUAL(msg.getHeaders().getAsInt64(REPLICATION_EVENT_SEQNO), (i+1)); + BOOST_CHECK_EQUAL((boost::format("%1%_%2%") % "Message" % (i+1)).str(), msg.getData()); + } +} + + +QPID_AUTO_TEST_SUITE_END() |