diff options
| author | Gordon Sim <gsim@apache.org> | 2011-02-10 10:12:41 +0000 |
|---|---|---|
| committer | Gordon Sim <gsim@apache.org> | 2011-02-10 10:12:41 +0000 |
| commit | 8b8d70e010a2999ad5dd1590d41eb35d8091296a (patch) | |
| tree | ae91e61fba3777fe9af4fb447d6b91387cbbc8db /qpid/cpp/src/qpid/cluster | |
| parent | 071d6276042c05be0f24b8e8bce205ae49d0480a (diff) | |
| download | qpid-python-8b8d70e010a2999ad5dd1590d41eb35d8091296a.tar.gz | |
QPID-529: Priority queue implementation
QPID-2104: LVQ enhancement
These both required some refactoring of the Queue class to allow cleaner implementation of different types of behaviour. The in-memory storage of messages is now abstracted out behind an interface specified by qpid::broker::Messages which qpid::broker::Queue uses. Different implementations of that are available for the standard FIFO queue, priority queues and LVQ (I have also separated out the 'legacy' implementation of LVQ from the new version driven by QPID-2104).
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1069322 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/cpp/src/qpid/cluster')
| -rw-r--r-- | qpid/cpp/src/qpid/cluster/Connection.cpp | 8 | ||||
| -rw-r--r-- | qpid/cpp/src/qpid/cluster/Connection.h | 1 | ||||
| -rw-r--r-- | qpid/cpp/src/qpid/cluster/UpdateClient.cpp | 5 |
3 files changed, 14 insertions, 0 deletions
diff --git a/qpid/cpp/src/qpid/cluster/Connection.cpp b/qpid/cpp/src/qpid/cluster/Connection.cpp index c7689577a7..0582945a9c 100644 --- a/qpid/cpp/src/qpid/cluster/Connection.cpp +++ b/qpid/cpp/src/qpid/cluster/Connection.cpp @@ -32,6 +32,7 @@ #include "qpid/broker/RecoveredEnqueue.h" #include "qpid/broker/RecoveredDequeue.h" #include "qpid/broker/Exchange.h" +#include "qpid/broker/Fairshare.h" #include "qpid/broker/Link.h" #include "qpid/broker/Bridge.h" #include "qpid/broker/Queue.h" @@ -548,6 +549,13 @@ void Connection::queuePosition(const string& qname, const SequenceNumber& positi findQueue(qname)->setPosition(position); } +void Connection::queueFairshareState(const std::string& qname, const uint8_t priority, const uint8_t count) +{ + if (!qpid::broker::Fairshare::setState(findQueue(qname)->getMessages(), priority, count)) { + QPID_LOG(error, "Failed to set fair share state on queue " << qname << "; this will result in inconsistencies."); + } +} + void Connection::expiryId(uint64_t id) { cluster.getExpiryPolicy().setId(id); } diff --git a/qpid/cpp/src/qpid/cluster/Connection.h b/qpid/cpp/src/qpid/cluster/Connection.h index d90cdd898b..7ee85bf1aa 100644 --- a/qpid/cpp/src/qpid/cluster/Connection.h +++ b/qpid/cpp/src/qpid/cluster/Connection.h @@ -152,6 +152,7 @@ class Connection : uint32_t credit); void queuePosition(const std::string&, const framing::SequenceNumber&); + void queueFairshareState(const std::string&, const uint8_t priority, const uint8_t count); void expiryId(uint64_t); void txStart(); diff --git a/qpid/cpp/src/qpid/cluster/UpdateClient.cpp b/qpid/cpp/src/qpid/cluster/UpdateClient.cpp index 4f6488a28a..8f751add9b 100644 --- a/qpid/cpp/src/qpid/cluster/UpdateClient.cpp +++ b/qpid/cpp/src/qpid/cluster/UpdateClient.cpp @@ -32,6 +32,7 @@ #include "qpid/client/ConnectionImpl.h" #include "qpid/client/Future.h" #include "qpid/broker/Broker.h" +#include "qpid/broker/Fairshare.h" #include "qpid/broker/Queue.h" #include "qpid/broker/QueueRegistry.h" #include "qpid/broker/LinkRegistry.h" @@ -352,6 +353,10 @@ void UpdateClient::updateQueue(client::AsyncSession& s, const boost::shared_ptr< q->eachMessage(boost::bind(&MessageUpdater::updateQueuedMessage, &updater, _1)); q->eachBinding(boost::bind(&UpdateClient::updateBinding, this, s, q->getName(), _1)); ClusterConnectionProxy(s).queuePosition(q->getName(), q->getPosition()); + uint priority, count; + if (qpid::broker::Fairshare::getState(q->getMessages(), priority, count)) { + ClusterConnectionProxy(s).queueFairshareState(q->getName(), priority, count); + } } void UpdateClient::updateExclusiveQueue(const boost::shared_ptr<broker::Queue>& q) { |