From d46a8ddf5964a965d7c6d0693e5df34db5c4d934 Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Mon, 28 Feb 2011 13:57:13 +0000 Subject: QPID-3087: Fixes to store interaction changes * don't create queue on recovery * ensure laternate exchange is set before creating store record for queue git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1075331 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qpid/broker/Broker.cpp | 7 +------ qpid/cpp/src/qpid/broker/Queue.cpp | 19 +++++++++++-------- qpid/cpp/src/qpid/broker/Queue.h | 13 ++++++++++--- qpid/cpp/src/qpid/broker/QueueRegistry.cpp | 21 ++++++++++++++++++--- qpid/cpp/src/qpid/broker/QueueRegistry.h | 5 ++++- qpid/cpp/src/tests/QueueTest.cpp | 8 ++++---- 6 files changed, 48 insertions(+), 25 deletions(-) diff --git a/qpid/cpp/src/qpid/broker/Broker.cpp b/qpid/cpp/src/qpid/broker/Broker.cpp index 4e7e78c961..695943854d 100644 --- a/qpid/cpp/src/qpid/broker/Broker.cpp +++ b/qpid/cpp/src/qpid/broker/Broker.cpp @@ -790,13 +790,8 @@ std::pair, bool> Broker::createQueue( if (!alternate) throw framing::NotFoundException(QPID_MSG("Alternate exchange does not exist: " << alternateExchange)); } - std::pair result = queues.declare(name, durable, autodelete, owner, arguments); + std::pair result = queues.declare(name, durable, autodelete, owner, alternate, arguments); if (result.second) { - if (alternate) { - result.first->setAlternateExchange(alternate); - alternate->incAlternateUsers(); - } - //add default binding: result.first->bind(exchanges.getDefault(), name); diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp index d18b0fcda3..09b8256b54 100644 --- a/qpid/cpp/src/qpid/broker/Queue.cpp +++ b/qpid/cpp/src/qpid/broker/Queue.cpp @@ -725,7 +725,7 @@ void Queue::create(const FieldTable& _settings) if (store) { store->create(*this, _settings); } - configure(_settings); + configureImpl(_settings); } @@ -750,9 +750,14 @@ int getIntegerSetting(const qpid::framing::FieldTable& settings, const std::stri } } -void Queue::configure(const FieldTable& _settings, bool recovering) +void Queue::configure(const FieldTable& _settings) { + settings = _settings; + configureImpl(settings); +} +void Queue::configureImpl(const FieldTable& _settings) +{ eventMode = _settings.getAsInt(qpidQueueEventGeneration); if (eventMode && broker) { broker->getQueueEvents().observe(*this, eventMode == ENQUEUE_ONLY); @@ -819,9 +824,6 @@ void Queue::configure(const FieldTable& _settings, bool recovering) mgmtObject->set_arguments(ManagementAgent::toMap(_settings)); } - if ( isDurable() && ! getPersistenceId() && ! recovering ) - store->create(*this, _settings); - QueueFlowLimit::observe(*this, _settings); } @@ -919,9 +921,10 @@ Queue::shared_ptr Queue::decode ( QueueRegistry& queues, Buffer& buffer, bool re { string name; buffer.getShortString(name); - std::pair result = queues.declare(name, true); - buffer.get(result.first->settings); - result.first->configure(result.first->settings, recovering ); + FieldTable settings; + buffer.get(settings); + boost::shared_ptr alternate; + std::pair result = queues.declare(name, true, false, 0, alternate, settings, recovering); if (result.first->policy.get() && buffer.available() >= result.first->policy->encodedSize()) { buffer.get ( *(result.first->policy) ); } diff --git a/qpid/cpp/src/qpid/broker/Queue.h b/qpid/cpp/src/qpid/broker/Queue.h index 331c9eaa4e..7597ec74ce 100644 --- a/qpid/cpp/src/qpid/broker/Queue.h +++ b/qpid/cpp/src/qpid/broker/Queue.h @@ -149,6 +149,7 @@ class Queue : public boost::enable_shared_from_this, QueuedMessage getFront(); void forcePersistent(QueuedMessage& msg); int getEventMode(); + void configureImpl(const qpid::framing::FieldTable& settings); inline void mgntEnqStats(const boost::intrusive_ptr& msg) { @@ -192,11 +193,17 @@ class Queue : public boost::enable_shared_from_this, QPID_BROKER_EXTERN bool dispatch(Consumer::shared_ptr); + /** + * Used to configure a new queue and create a persistent record + * for it in store if required. + */ void create(const qpid::framing::FieldTable& settings); - // "recovering" means we are doing a MessageStore recovery. - QPID_BROKER_EXTERN void configure(const qpid::framing::FieldTable& settings, - bool recovering = false); + /** + * Used to reconfigure a recovered queue (does not create + * persistent record in store). + */ + QPID_BROKER_EXTERN void configure(const qpid::framing::FieldTable& settings); void destroyed(); QPID_BROKER_EXTERN void bound(const std::string& exchange, const std::string& key, diff --git a/qpid/cpp/src/qpid/broker/QueueRegistry.cpp b/qpid/cpp/src/qpid/broker/QueueRegistry.cpp index c44b11e4bf..135a3543d9 100644 --- a/qpid/cpp/src/qpid/broker/QueueRegistry.cpp +++ b/qpid/cpp/src/qpid/broker/QueueRegistry.cpp @@ -21,6 +21,7 @@ #include "qpid/broker/Queue.h" #include "qpid/broker/QueueRegistry.h" #include "qpid/broker/QueueEvents.h" +#include "qpid/broker/Exchange.h" #include "qpid/log/Statement.h" #include #include @@ -37,7 +38,12 @@ QueueRegistry::~QueueRegistry(){} std::pair QueueRegistry::declare(const string& declareName, bool durable, bool autoDelete, const OwnershipToken* owner, - const qpid::framing::FieldTable& arguments) + boost::shared_ptr alternate, + const qpid::framing::FieldTable& arguments, + bool recovering/*true if this declare is a + result of recovering queue + definition from persistente + record*/) { RWlock::ScopedWlock locker(lock); string name = declareName.empty() ? generateName() : declareName; @@ -46,8 +52,17 @@ QueueRegistry::declare(const string& declareName, bool durable, if (i == queues.end()) { Queue::shared_ptr queue(new Queue(name, autoDelete, durable ? store : 0, owner, parent, broker)); - //apply settings & create persistent record if required - queue->create(arguments); + if (alternate) { + queue->setAlternateExchange(alternate);//need to do this *before* create + alternate->incAlternateUsers(); + } + if (!recovering) { + //apply settings & create persistent record if required + queue->create(arguments); + } else { + //i.e. recovering a queue for which we already have a persistent record + queue->configure(arguments); + } queues[name] = queue; if (lastNode) queue->setLastNodeFailure(); diff --git a/qpid/cpp/src/qpid/broker/QueueRegistry.h b/qpid/cpp/src/qpid/broker/QueueRegistry.h index 90ee924ba4..8a32a64f05 100644 --- a/qpid/cpp/src/qpid/broker/QueueRegistry.h +++ b/qpid/cpp/src/qpid/broker/QueueRegistry.h @@ -35,6 +35,7 @@ namespace broker { class Queue; class QueueEvents; +class Exchange; class OwnershipToken; class Broker; class MessageStore; @@ -62,7 +63,9 @@ class QueueRegistry { bool durable = false, bool autodelete = false, const OwnershipToken* owner = 0, - const qpid::framing::FieldTable& args = framing::FieldTable()); + boost::shared_ptr alternateExchange = boost::shared_ptr(), + const qpid::framing::FieldTable& args = framing::FieldTable(), + bool recovering = false); /** * Destroy the named queue. diff --git a/qpid/cpp/src/tests/QueueTest.cpp b/qpid/cpp/src/tests/QueueTest.cpp index e4e9897195..fd30a98ac0 100644 --- a/qpid/cpp/src/tests/QueueTest.cpp +++ b/qpid/cpp/src/tests/QueueTest.cpp @@ -637,7 +637,7 @@ QPID_AUTO_TEST_CASE(testLVQRecover){ Queue::shared_ptr queue1(new Queue("my-queue", true, &testStore)); intrusive_ptr received; - queue1->configure(args); + queue1->create(args); intrusive_ptr msg1 = create_message("e", "A"); intrusive_ptr msg2 = create_message("e", "A"); @@ -709,9 +709,9 @@ QPID_AUTO_TEST_CASE(testMultiQueueLastNode){ args.setPersistLastNode(); Queue::shared_ptr queue1(new Queue("queue1", true, &testStore )); - queue1->configure(args); + queue1->create(args); Queue::shared_ptr queue2(new Queue("queue2", true, &testStore )); - queue2->configure(args); + queue2->create(args); intrusive_ptr msg1 = create_message("e", "A"); @@ -797,7 +797,7 @@ not requeued to the store. Queue::shared_ptr queue1(new Queue("my-queue", true, &testStore)); intrusive_ptr received; - queue1->configure(args); + queue1->create(args); // check requeue 1 intrusive_ptr msg1 = create_message("e", "C"); -- cgit v1.2.1