diff options
| author | Gordon Sim <gsim@apache.org> | 2011-02-28 13:57:13 +0000 |
|---|---|---|
| committer | Gordon Sim <gsim@apache.org> | 2011-02-28 13:57:13 +0000 |
| commit | e37fecdfd9a01858f5e5d0d4e20d53b8ceb50801 (patch) | |
| tree | 59d5e5146d1c2ba7381925605773405ab8a74534 /cpp/src/qpid | |
| parent | 3a7a8d0fea6995aeee3d0e3a982b240ae7ef79a2 (diff) | |
| download | qpid-python-e37fecdfd9a01858f5e5d0d4e20d53b8ceb50801.tar.gz | |
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/qpid@1075331 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'cpp/src/qpid')
| -rw-r--r-- | cpp/src/qpid/broker/Broker.cpp | 7 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/Queue.cpp | 19 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/Queue.h | 13 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/QueueRegistry.cpp | 21 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/QueueRegistry.h | 5 |
5 files changed, 44 insertions, 21 deletions
diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp index 4e7e78c961..695943854d 100644 --- a/cpp/src/qpid/broker/Broker.cpp +++ b/cpp/src/qpid/broker/Broker.cpp @@ -790,13 +790,8 @@ std::pair<boost::shared_ptr<Queue>, bool> Broker::createQueue( if (!alternate) throw framing::NotFoundException(QPID_MSG("Alternate exchange does not exist: " << alternateExchange)); } - std::pair<Queue::shared_ptr, bool> result = queues.declare(name, durable, autodelete, owner, arguments); + std::pair<Queue::shared_ptr, bool> 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/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp index d18b0fcda3..09b8256b54 100644 --- a/cpp/src/qpid/broker/Queue.cpp +++ b/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<Queue::shared_ptr, bool> 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<Exchange> alternate; + std::pair<Queue::shared_ptr, bool> 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/cpp/src/qpid/broker/Queue.h b/cpp/src/qpid/broker/Queue.h index 331c9eaa4e..7597ec74ce 100644 --- a/cpp/src/qpid/broker/Queue.h +++ b/cpp/src/qpid/broker/Queue.h @@ -149,6 +149,7 @@ class Queue : public boost::enable_shared_from_this<Queue>, QueuedMessage getFront(); void forcePersistent(QueuedMessage& msg); int getEventMode(); + void configureImpl(const qpid::framing::FieldTable& settings); inline void mgntEnqStats(const boost::intrusive_ptr<Message>& msg) { @@ -192,11 +193,17 @@ class Queue : public boost::enable_shared_from_this<Queue>, 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/cpp/src/qpid/broker/QueueRegistry.cpp b/cpp/src/qpid/broker/QueueRegistry.cpp index c44b11e4bf..135a3543d9 100644 --- a/cpp/src/qpid/broker/QueueRegistry.cpp +++ b/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 <sstream> #include <assert.h> @@ -37,7 +38,12 @@ QueueRegistry::~QueueRegistry(){} std::pair<Queue::shared_ptr, bool> QueueRegistry::declare(const string& declareName, bool durable, bool autoDelete, const OwnershipToken* owner, - const qpid::framing::FieldTable& arguments) + boost::shared_ptr<Exchange> 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/cpp/src/qpid/broker/QueueRegistry.h b/cpp/src/qpid/broker/QueueRegistry.h index 90ee924ba4..8a32a64f05 100644 --- a/cpp/src/qpid/broker/QueueRegistry.h +++ b/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<Exchange> alternateExchange = boost::shared_ptr<Exchange>(), + const qpid::framing::FieldTable& args = framing::FieldTable(), + bool recovering = false); /** * Destroy the named queue. |
