You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2011/11/09 22:31:08 UTC
svn commit: r1199975 - in /qpid/branches/qpid-3603/qpid: cpp/src/
cpp/src/qpid/broker/ cpp/src/qpid/client/amqp0_10/ cpp/src/tests/
python/qpid/tests/messaging/ tests/src/py/qpid_tests/broker_0_10/
Author: aconway
Date: Wed Nov 9 21:31:07 2011
New Revision: 1199975
URL: http://svn.apache.org/viewvc?rev=1199975&view=rev
Log:
QPID-3603: Prototype of replicating browser.
Added:
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.cpp (with props)
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.h (with props)
Modified:
qpid/branches/qpid-3603/qpid/cpp/src/CMakeLists.txt
qpid/branches/qpid-3603/qpid/cpp/src/Makefile.am
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Bridge.cpp
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/DeliveryRecord.h
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/FifoDistributor.cpp
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/LegacyLVQ.cpp
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/LegacyLVQ.h
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageDeque.cpp
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageDeque.h
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageMap.cpp
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageMap.h
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Messages.h
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/PriorityQueue.cpp
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/PriorityQueue.h
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Queue.cpp
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Queue.h
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueuedMessage.h
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/SemanticState.cpp
qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/SemanticState.h
qpid/branches/qpid-3603/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp
qpid/branches/qpid-3603/qpid/cpp/src/tests/QueueTest.cpp
qpid/branches/qpid-3603/qpid/python/qpid/tests/messaging/endpoints.py
qpid/branches/qpid-3603/qpid/tests/src/py/qpid_tests/broker_0_10/message.py
qpid/branches/qpid-3603/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py
Modified: qpid/branches/qpid-3603/qpid/cpp/src/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/CMakeLists.txt?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/CMakeLists.txt (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/CMakeLists.txt Wed Nov 9 21:31:07 2011
@@ -1005,6 +1005,7 @@ set (qpidbroker_SOURCES
qpid/broker/QueueListeners.cpp
qpid/broker/FifoDistributor.cpp
qpid/broker/MessageGroupManager.cpp
+ qpid/broker/QueueReplicator.cpp
qpid/broker/PersistableMessage.cpp
qpid/broker/Bridge.cpp
qpid/broker/Connection.cpp
Modified: qpid/branches/qpid-3603/qpid/cpp/src/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/Makefile.am?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/Makefile.am (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/Makefile.am Wed Nov 9 21:31:07 2011
@@ -620,6 +620,8 @@ libqpidbroker_la_SOURCES = \
qpid/broker/QueuedMessage.h \
qpid/broker/QueueFlowLimit.h \
qpid/broker/QueueFlowLimit.cpp \
+ qpid/broker/QueueReplicator.h \
+ qpid/broker/QueueReplicator.cpp \
qpid/broker/RateFlowcontrol.h \
qpid/broker/RecoverableConfig.h \
qpid/broker/RecoverableExchange.h \
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Bridge.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Bridge.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Bridge.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Bridge.cpp Wed Nov 9 21:31:07 2011
@@ -24,6 +24,7 @@
#include "qpid/broker/Connection.h"
#include "qpid/broker/Link.h"
#include "qpid/broker/LinkRegistry.h"
+#include "qpid/broker/QueueReplicator.h"
#include "qpid/broker/SessionState.h"
#include "qpid/management/ManagementAgent.h"
@@ -96,8 +97,11 @@ void Bridge::create(Connection& c)
}
if (args.i_srcIsLocal) sessionHandler.getSession()->disableReceiverTracking();
- if (args.i_srcIsQueue) {
- peer->getMessage().subscribe(args.i_src, args.i_dest, args.i_sync ? 0 : 1, 0, false, "", 0, options);
+ if (args.i_srcIsQueue) {
+ //TODO: something other than this which is nasty...
+ bool isReplicatingLink = QueueReplicator::initReplicationSettings(args.i_dest, link->getBroker()->getQueues(), options);
+
+ peer->getMessage().subscribe(args.i_src, args.i_dest, args.i_sync ? 0 : 1, isReplicatingLink ? 1 : 0, false, "", 0, options);
peer->getMessage().flow(args.i_dest, 0, 0xFFFFFFFF);
peer->getMessage().flow(args.i_dest, 1, 0xFFFFFFFF);
QPID_LOG(debug, "Activated route from queue " << args.i_src << " to " << args.i_dest);
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp Wed Nov 9 21:31:07 2011
@@ -37,7 +37,7 @@ DeliveryRecord::DeliveryRecord(const Que
bool _acquired,
bool accepted,
bool _windowing,
- uint32_t _credit) : msg(_msg),
+ uint32_t _credit, bool _delayedCompletion) : msg(_msg),
queue(_queue),
tag(_tag),
acquired(_acquired),
@@ -46,7 +46,8 @@ DeliveryRecord::DeliveryRecord(const Que
completed(false),
ended(accepted && acquired),
windowing(_windowing),
- credit(msg.payload ? msg.payload->getRequiredCredit() : _credit)
+ credit(msg.payload ? msg.payload->getRequiredCredit() : _credit),
+ delayedCompletion(_delayedCompletion)
{}
bool DeliveryRecord::setEnded()
@@ -111,8 +112,14 @@ void DeliveryRecord::complete() {
}
bool DeliveryRecord::accept(TransactionContext* ctxt) {
- if (acquired && !ended) {
- queue->dequeue(ctxt, msg);
+ if (!ended) {
+ if (acquired) {
+ queue->dequeue(ctxt, msg);
+ } else if (delayedCompletion) {
+ //TODO: this is a nasty way to do this; change it
+ msg.payload->getIngressCompletion().finishCompleter();
+ QPID_LOG(debug, "Completed " << msg.payload.get());
+ }
setEnded();
QPID_LOG(debug, "Accepted " << id);
}
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/DeliveryRecord.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/DeliveryRecord.h?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/DeliveryRecord.h (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/DeliveryRecord.h Wed Nov 9 21:31:07 2011
@@ -63,6 +63,7 @@ class DeliveryRecord
* after that).
*/
uint32_t credit;
+ bool delayedCompletion;
public:
QPID_BROKER_EXTERN DeliveryRecord(const QueuedMessage& msg,
@@ -71,7 +72,8 @@ class DeliveryRecord
bool acquired,
bool accepted,
bool windowing,
- uint32_t credit=0 // Only used if msg is empty.
+ uint32_t credit=0, // Only used if msg is empty.
+ bool delayedCompletion=false
);
bool coveredBy(const framing::SequenceSet* const range) const { return range->contains(id); }
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/FifoDistributor.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/FifoDistributor.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/FifoDistributor.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/FifoDistributor.cpp Wed Nov 9 21:31:07 2011
@@ -30,11 +30,7 @@ FifoDistributor::FifoDistributor(Message
bool FifoDistributor::nextConsumableMessage( Consumer::shared_ptr&, QueuedMessage& next )
{
- if (!messages.empty()) {
- next = messages.front(); // by default, consume oldest msg
- return true;
- }
- return false;
+ return messages.consume(next);
}
bool FifoDistributor::allocate(const std::string&, const QueuedMessage& )
@@ -46,9 +42,7 @@ bool FifoDistributor::allocate(const std
bool FifoDistributor::nextBrowsableMessage( Consumer::shared_ptr& c, QueuedMessage& next )
{
- if (!messages.empty() && messages.next(c->position, next))
- return true;
- return false;
+ return messages.browse(c->position, next, false);
}
void FifoDistributor::query(qpid::types::Variant::Map&) const
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/LegacyLVQ.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/LegacyLVQ.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/LegacyLVQ.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/LegacyLVQ.cpp Wed Nov 9 21:31:07 2011
@@ -32,7 +32,7 @@ void LegacyLVQ::setNoBrowse(bool b)
noBrowse = b;
}
-bool LegacyLVQ::remove(const framing::SequenceNumber& position, QueuedMessage& message)
+bool LegacyLVQ::acquire(const framing::SequenceNumber& position, QueuedMessage& message)
{
Ordering::iterator i = messages.find(position);
if (i != messages.end() && i->second.payload == message.payload) {
@@ -44,9 +44,9 @@ bool LegacyLVQ::remove(const framing::Se
}
}
-bool LegacyLVQ::next(const framing::SequenceNumber& position, QueuedMessage& message)
+bool LegacyLVQ::browse(const framing::SequenceNumber& position, QueuedMessage& message, bool unacquired)
{
- if (MessageMap::next(position, message)) {
+ if (MessageMap::browse(position, message, unacquired)) {
if (!noBrowse) index.erase(getKey(message));
return true;
} else {
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/LegacyLVQ.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/LegacyLVQ.h?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/LegacyLVQ.h (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/LegacyLVQ.h Wed Nov 9 21:31:07 2011
@@ -40,8 +40,8 @@ class LegacyLVQ : public MessageMap
{
public:
LegacyLVQ(const std::string& key, bool noBrowse = false, Broker* broker = 0);
- bool remove(const framing::SequenceNumber&, QueuedMessage&);
- bool next(const framing::SequenceNumber&, QueuedMessage&);
+ bool acquire(const framing::SequenceNumber&, QueuedMessage&);
+ bool browse(const framing::SequenceNumber&, QueuedMessage&, bool);
bool push(const QueuedMessage& added, QueuedMessage& removed);
void removeIf(Predicate);
void setNoBrowse(bool);
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageDeque.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageDeque.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageDeque.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageDeque.cpp Wed Nov 9 21:31:07 2011
@@ -20,121 +20,155 @@
*/
#include "qpid/broker/MessageDeque.h"
#include "qpid/broker/QueuedMessage.h"
+#include "qpid/log/Statement.h"
namespace qpid {
namespace broker {
-size_t MessageDeque::size()
-{
- return messages.size();
-}
-
-bool MessageDeque::empty()
-{
- return messages.empty();
-}
+MessageDeque::MessageDeque() : available(0), head(0) {}
-void MessageDeque::reinsert(const QueuedMessage& message)
+size_t MessageDeque::index(const framing::SequenceNumber& position)
{
- messages.insert(lower_bound(messages.begin(), messages.end(), message), message);
-}
-
-MessageDeque::Deque::iterator MessageDeque::seek(const framing::SequenceNumber& position)
-{
- if (!messages.empty()) {
- QueuedMessage comp;
- comp.position = position;
- unsigned long diff = position.getValue() - messages.front().position.getValue();
- long maxEnd = diff < messages.size()? diff : messages.size();
- return lower_bound(messages.begin(),messages.begin()+maxEnd,comp);
- } else {
- return messages.end();
- }
+ //assuming a monotonic sequence, with no messages removed except
+ //from the ends of the deque, we can use the position to determin
+ //an index into the deque
+ if (messages.empty() || position < messages.front().position) return 0;
+ return position - messages.front().position;
}
-bool MessageDeque::find(const framing::SequenceNumber& position, QueuedMessage& message, bool remove)
+bool MessageDeque::deleted(const QueuedMessage& m)
{
- Deque::iterator i = seek(position);
- if (i != messages.end() && i->position == position) {
- message = *i;
- if (remove) messages.erase(i);
+ size_t i = index(m.position);
+ if (i < messages.size()) {
+ messages[i].status = QueuedMessage::DELETED;
+ clean();
return true;
} else {
return false;
}
}
-bool MessageDeque::remove(const framing::SequenceNumber& position, QueuedMessage& message)
+size_t MessageDeque::size()
{
- return find(position, message, true);
+ return available;
}
-bool MessageDeque::find(const framing::SequenceNumber& position, QueuedMessage& message)
+void MessageDeque::release(const QueuedMessage& message)
{
- return find(position, message, false);
+ size_t i = index(message.position);
+ if (i < messages.size()) {
+ QueuedMessage& m = messages[i];
+ if (m.status == QueuedMessage::ACQUIRED) {
+ if (head > i) head = i;
+ m.status = QueuedMessage::AVAILABLE;
+ ++available;
+ }
+ } else {
+ QPID_LOG(error, "Failed to release message at " << message.position << " " << message.payload->getFrames().getContent() << "; no such message (index=" << i << ", size=" << messages.size() << ")");
+ }
}
-bool MessageDeque::next(const framing::SequenceNumber& position, QueuedMessage& message)
+bool MessageDeque::acquire(const framing::SequenceNumber& position, QueuedMessage& message)
{
- if (messages.empty()) {
- return false;
- } else if (position < front().position) {
- message = front();
- return true;
- } else {
- Deque::iterator i = seek(position+1);
- if (i != messages.end()) {
- message = *i;
+ size_t i = index(position);
+ if (i < messages.size()) {
+ QueuedMessage& temp = messages[i];
+ if (temp.status == QueuedMessage::AVAILABLE) {
+ temp.status = QueuedMessage::ACQUIRED;
+ --available;
+ message = temp;
return true;
- } else {
- return false;
}
}
+ return false;
}
-QueuedMessage& MessageDeque::front()
+bool MessageDeque::find(const framing::SequenceNumber& position, QueuedMessage& message)
{
- return messages.front();
+ size_t i = index(position);
+ if (i < messages.size()) {
+ message = messages[i];
+ return true;
+ } else {
+ return false;
+ }
}
-void MessageDeque::pop()
+bool MessageDeque::browse(const framing::SequenceNumber& position, QueuedMessage& message, bool unacquired)
{
- if (!messages.empty()) {
- messages.pop_front();
+ //get first message that is greater than position
+ size_t i = index(position + 1);
+ while (i < messages.size()) {
+ QueuedMessage& m = messages[i++];
+ if (m.status == QueuedMessage::AVAILABLE || (!unacquired && m.status == QueuedMessage::ACQUIRED)) {
+ message = m;
+ return true;
+ }
}
+ return false;
}
-bool MessageDeque::pop(QueuedMessage& out)
+bool MessageDeque::consume(QueuedMessage& message)
{
- if (messages.empty()) {
- return false;
- } else {
- out = front();
- messages.pop_front();
- return true;
+ while (head < messages.size()) {
+ QueuedMessage& i = messages[head++];
+ if (i.status == QueuedMessage::AVAILABLE) {
+ i.status = QueuedMessage::ACQUIRED;
+ --available;
+ message = i;
+ return true;
+ }
}
+ return false;
}
bool MessageDeque::push(const QueuedMessage& added, QueuedMessage& /*not needed*/)
{
+ //add padding to prevent gaps in sequence, which break the index
+ //calculation (needed for queue replication)
+ while (messages.size() && (added.position - messages.back().position) > 1) {
+ QueuedMessage dummy;
+ dummy.position = messages.back().position + 1;
+ dummy.status = QueuedMessage::DELETED;
+ messages.push_back(dummy);
+ QPID_LOG(debug, "Adding padding at " << dummy.position << ", between " << messages.back().position << " and " << added.position);
+ }
messages.push_back(added);
+ messages.back().status = QueuedMessage::AVAILABLE;
+ if (head >= messages.size()) head = messages.size() - 1;
+ ++available;
return false;//adding a message never causes one to be removed for deque
}
+void MessageDeque::clean()
+{
+ while (messages.size() && messages.front().status == QueuedMessage::DELETED) {
+ messages.pop_front();
+ if (head) --head;
+ }
+}
+
void MessageDeque::foreach(Functor f)
{
- std::for_each(messages.begin(), messages.end(), f);
+ for (Deque::iterator i = messages.begin(); i != messages.end(); ++i) {
+ if (i->status == QueuedMessage::AVAILABLE) {
+ f(*i);
+ }
+ }
}
void MessageDeque::removeIf(Predicate p)
{
- for (Deque::iterator i = messages.begin(); i != messages.end();) {
- if (p(*i)) {
- i = messages.erase(i);
- } else {
- ++i;
+ for (Deque::iterator i = messages.begin(); i != messages.end(); ++i) {
+ if (i->status == QueuedMessage::AVAILABLE && p(*i)) {
+ //Use special status for this as messages are not yet
+ //dequeued, but should not be considered on the queue
+ //either (used for purging and moving)
+ i->status = QueuedMessage::REMOVED;
+ --available;
}
}
+ clean();
}
}} // namespace qpid::broker
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageDeque.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageDeque.h?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageDeque.h (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageDeque.h Wed Nov 9 21:31:07 2011
@@ -34,17 +34,14 @@ namespace broker {
class MessageDeque : public Messages
{
public:
+ MessageDeque();
size_t size();
- bool empty();
-
- void reinsert(const QueuedMessage&);
- bool remove(const framing::SequenceNumber&, QueuedMessage&);
+ bool deleted(const QueuedMessage&);
+ void release(const QueuedMessage&);
+ bool acquire(const framing::SequenceNumber&, QueuedMessage&);
bool find(const framing::SequenceNumber&, QueuedMessage&);
- bool next(const framing::SequenceNumber&, QueuedMessage&);
-
- QueuedMessage& front();
- void pop();
- bool pop(QueuedMessage&);
+ bool browse(const framing::SequenceNumber&, QueuedMessage&, bool);
+ bool consume(QueuedMessage&);
bool push(const QueuedMessage& added, QueuedMessage& removed);
void foreach(Functor);
@@ -53,9 +50,11 @@ class MessageDeque : public Messages
private:
typedef std::deque<QueuedMessage> Deque;
Deque messages;
+ size_t available;
+ size_t head;
- Deque::iterator seek(const framing::SequenceNumber&);
- bool find(const framing::SequenceNumber&, QueuedMessage&, bool remove);
+ size_t index(const framing::SequenceNumber&);
+ void clean();
};
}} // namespace qpid::broker
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp Wed Nov 9 21:31:07 2011
@@ -204,7 +204,7 @@ MessageGroupManager::~MessageGroupManage
}
bool MessageGroupManager::nextConsumableMessage( Consumer::shared_ptr& c, QueuedMessage& next )
{
- if (messages.empty())
+ if (!messages.size())
return false;
next.position = c->position;
@@ -216,15 +216,16 @@ bool MessageGroupManager::nextConsumable
}
}
- while (messages.next( next.position, next )) {
+ while (messages.browse( next.position, next, true )) {
GroupState& group = findGroup(next);
if (!group.owned()) {
- if (group.members.front() == next.position) { // only take from head!
+ //TODO: make acquire more efficient when we already have the message in question
+ if (group.members.front() == next.position && messages.acquire(next.position, next)) { // only take from head!
return true;
}
QPID_LOG(debug, "Skipping " << next.position << " since group " << group.group
<< "'s head message still pending. pos=" << group.members.front());
- } else if (group.owner == c->getName()) {
+ } else if (group.owner == c->getName() && messages.acquire(next.position, next)) {
return true;
}
}
@@ -249,9 +250,7 @@ bool MessageGroupManager::allocate(const
bool MessageGroupManager::nextBrowsableMessage( Consumer::shared_ptr& c, QueuedMessage& next )
{
// browse: allow access to any available msg, regardless of group ownership (?ok?)
- if (!messages.empty() && messages.next(c->position, next))
- return true;
- return false;
+ return messages.browse(c->position, next, false);
}
void MessageGroupManager::query(qpid::types::Variant::Map& status) const
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageMap.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageMap.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageMap.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageMap.cpp Wed Nov 9 21:31:07 2011
@@ -27,6 +27,8 @@ namespace {
const std::string EMPTY;
}
+bool MessageMap::deleted(const QueuedMessage&) { return true; }
+
std::string MessageMap::getKey(const QueuedMessage& message)
{
const framing::FieldTable* ft = message.payload->getApplicationHeaders();
@@ -44,7 +46,7 @@ bool MessageMap::empty()
return messages.empty();
}
-void MessageMap::reinsert(const QueuedMessage& message)
+void MessageMap::release(const QueuedMessage& message)
{
std::string key = getKey(message);
Index::iterator i = index.find(key);
@@ -54,7 +56,7 @@ void MessageMap::reinsert(const QueuedMe
} //else message has already been replaced
}
-bool MessageMap::remove(const framing::SequenceNumber& position, QueuedMessage& message)
+bool MessageMap::acquire(const framing::SequenceNumber& position, QueuedMessage& message)
{
Ordering::iterator i = messages.find(position);
if (i != messages.end()) {
@@ -77,38 +79,22 @@ bool MessageMap::find(const framing::Seq
}
}
-bool MessageMap::next(const framing::SequenceNumber& position, QueuedMessage& message)
+bool MessageMap::browse(const framing::SequenceNumber& position, QueuedMessage& message, bool)
{
- if (!messages.empty() && position < front().position) {
- message = front();
+ Ordering::iterator i = messages.lower_bound(position+1);
+ if (i != messages.end()) {
+ message = i->second;
return true;
} else {
- Ordering::iterator i = messages.lower_bound(position+1);
- if (i != messages.end()) {
- message = i->second;
- return true;
- } else {
- return false;
- }
+ return false;
}
}
-QueuedMessage& MessageMap::front()
-{
- return messages.begin()->second;
-}
-
-void MessageMap::pop()
-{
- QueuedMessage dummy;
- pop(dummy);
-}
-
-bool MessageMap::pop(QueuedMessage& out)
+bool MessageMap::consume(QueuedMessage& message)
{
Ordering::iterator i = messages.begin();
if (i != messages.end()) {
- out = i->second;
+ message = i->second;
erase(i);
return true;
} else {
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageMap.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageMap.h?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageMap.h (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/MessageMap.h Wed Nov 9 21:31:07 2011
@@ -43,14 +43,12 @@ class MessageMap : public Messages
size_t size();
bool empty();
- void reinsert(const QueuedMessage&);
- virtual bool remove(const framing::SequenceNumber&, QueuedMessage&);
+ bool deleted(const QueuedMessage&);
+ void release(const QueuedMessage&);
+ virtual bool acquire(const framing::SequenceNumber&, QueuedMessage&);
bool find(const framing::SequenceNumber&, QueuedMessage&);
- virtual bool next(const framing::SequenceNumber&, QueuedMessage&);
-
- QueuedMessage& front();
- void pop();
- bool pop(QueuedMessage&);
+ virtual bool browse(const framing::SequenceNumber&, QueuedMessage&, bool);
+ bool consume(QueuedMessage&);
virtual bool push(const QueuedMessage& added, QueuedMessage& removed);
void foreach(Functor);
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Messages.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Messages.h?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Messages.h (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Messages.h Wed Nov 9 21:31:07 2011
@@ -46,22 +46,21 @@ class Messages
* @return the number of messages available for delivery.
*/
virtual size_t size() = 0;
+
/**
- * @return true if there are no messages for delivery, false otherwise
+ * Called when a message is deleted from the queue.
*/
- virtual bool empty() = 0;
-
+ virtual bool deleted(const QueuedMessage&) = 0;
/**
- * Re-inserts a message back into its original position - used
- * when requeing released messages.
+ * Releases an acquired message, making it available again.
*/
- virtual void reinsert(const QueuedMessage&) = 0;
+ virtual void release(const QueuedMessage&) = 0;
/**
- * Remove the message at the specified position, returning true if
- * found, false otherwise. The removed message is passed back via
- * the second parameter.
+ * Acquire the message at the specified position, returning true
+ * if found, false otherwise. The acquired message is passed back
+ * via the second parameter.
*/
- virtual bool remove(const framing::SequenceNumber&, QueuedMessage&) = 0;
+ virtual bool acquire(const framing::SequenceNumber&, QueuedMessage&) = 0;
/**
* Find the message at the specified position, returning true if
* found, false otherwise. The matched message is passed back via
@@ -69,30 +68,22 @@ class Messages
*/
virtual bool find(const framing::SequenceNumber&, QueuedMessage&) = 0;
/**
- * Return the next message to be given to a browsing subscrption
- * that has reached the specified poisition. The next messages is
- * passed back via the second parameter.
+ * Retrieve the next message to be given to a browsing
+ * subscription that has reached the specified position. The next
+ * message is passed back via the second parameter.
+ *
+ * @param unacquired, if true, will only browse unacquired messages
*
* @return true if there is another message, false otherwise.
*/
- virtual bool next(const framing::SequenceNumber&, QueuedMessage&) = 0;
+ virtual bool browse(const framing::SequenceNumber&, QueuedMessage&, bool unacquired) = 0;
/**
- * Note: Caller is responsible for ensuring that there is a front
- * (e.g. empty() returns false)
+ * Retrieve the next message available for a consuming
+ * subscription.
*
- * @return the next message to be delivered
- */
- virtual QueuedMessage& front() = 0;
- /**
- * Removes the front message
- */
- virtual void pop() = 0;
- /**
- * @return true if there is a mesage to be delivered - in which
- * case that message will be returned via the parameter and
- * removed - otherwise false.
+ * @return true if there is such a message, false otherwise.
*/
- virtual bool pop(QueuedMessage&) = 0;
+ virtual bool consume(QueuedMessage&) = 0;
/**
* Pushes a message to the back of the 'queue'. For some types of
* queue this may cause another message to be removed; if that is
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/PriorityQueue.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/PriorityQueue.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/PriorityQueue.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/PriorityQueue.cpp Wed Nov 9 21:31:07 2011
@@ -32,6 +32,8 @@ PriorityQueue::PriorityQueue(int l) :
messages(levels, Deque()),
frontLevel(0), haveFront(false), cached(false) {}
+bool PriorityQueue::deleted(const QueuedMessage&) { return true; }
+
size_t PriorityQueue::size()
{
size_t total(0);
@@ -41,15 +43,7 @@ size_t PriorityQueue::size()
return total;
}
-bool PriorityQueue::empty()
-{
- for (int i = 0; i < levels; ++i) {
- if (!messages[i].empty()) return false;
- }
- return true;
-}
-
-void PriorityQueue::reinsert(const QueuedMessage& message)
+void PriorityQueue::release(const QueuedMessage& message)
{
uint p = getPriorityLevel(message);
messages[p].insert(lower_bound(messages[p].begin(), messages[p].end(), message), message);
@@ -78,7 +72,7 @@ bool PriorityQueue::find(const framing::
return false;
}
-bool PriorityQueue::remove(const framing::SequenceNumber& position, QueuedMessage& message)
+bool PriorityQueue::acquire(const framing::SequenceNumber& position, QueuedMessage& message)
{
return find(position, message, true);
}
@@ -88,7 +82,7 @@ bool PriorityQueue::find(const framing::
return find(position, message, false);
}
-bool PriorityQueue::next(const framing::SequenceNumber& position, QueuedMessage& message)
+bool PriorityQueue::browse(const framing::SequenceNumber& position, QueuedMessage& message, bool)
{
QueuedMessage match;
match.position = position+1;
@@ -112,16 +106,7 @@ bool PriorityQueue::next(const framing::
return found;
}
-QueuedMessage& PriorityQueue::front()
-{
- if (checkFront()) {
- return messages[frontLevel].front();
- } else {
- throw qpid::framing::InternalErrorException(QPID_MSG("No message available"));
- }
-}
-
-bool PriorityQueue::pop(QueuedMessage& message)
+bool PriorityQueue::consume(QueuedMessage& message)
{
if (checkFront()) {
message = messages[frontLevel].front();
@@ -133,12 +118,6 @@ bool PriorityQueue::pop(QueuedMessage& m
}
}
-void PriorityQueue::pop()
-{
- QueuedMessage dummy;
- pop(dummy);
-}
-
bool PriorityQueue::push(const QueuedMessage& added, QueuedMessage& /*not needed*/)
{
messages[getPriorityLevel(added)].push_back(added);
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/PriorityQueue.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/PriorityQueue.h?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/PriorityQueue.h (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/PriorityQueue.h Wed Nov 9 21:31:07 2011
@@ -40,16 +40,13 @@ class PriorityQueue : public Messages
PriorityQueue(int levels);
virtual ~PriorityQueue() {}
size_t size();
- bool empty();
- void reinsert(const QueuedMessage&);
- bool remove(const framing::SequenceNumber&, QueuedMessage&);
+ bool deleted(const QueuedMessage&);
+ void release(const QueuedMessage&);
+ bool acquire(const framing::SequenceNumber&, QueuedMessage&);
bool find(const framing::SequenceNumber&, QueuedMessage&);
- bool next(const framing::SequenceNumber&, QueuedMessage&);
-
- QueuedMessage& front();
- void pop();
- bool pop(QueuedMessage&);
+ bool browse(const framing::SequenceNumber&, QueuedMessage&, bool);
+ bool consume(QueuedMessage&);
bool push(const QueuedMessage& added, QueuedMessage& removed);
void foreach(Functor);
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Queue.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Queue.cpp Wed Nov 9 21:31:07 2011
@@ -213,7 +213,7 @@ void Queue::requeue(const QueuedMessage&
{
Mutex::ScopedLock locker(messageLock);
if (!isEnqueued(msg)) return;
- messages->reinsert(msg);
+ messages->release(msg);
listeners.populate(copy);
// for persistLastNode - don't force a message twice to disk, but force it if no force before
@@ -296,46 +296,40 @@ bool Queue::getNextMessage(QueuedMessage
Queue::ConsumeCode Queue::consumeNextMessage(QueuedMessage& m, Consumer::shared_ptr& c)
{
+ QueuedMessage msg;
while (true) {
Mutex::ScopedLock locker(messageLock);
- QueuedMessage msg;
-
- if (!allocator->nextConsumableMessage(c, msg)) { // no next available
- QPID_LOG(debug, "No messages available to dispatch to consumer " <<
- c->getName() << " on queue '" << name << "'");
- listeners.addListener(c);
- return NO_MESSAGES;
- }
-
- if (msg.payload->hasExpired()) {
- QPID_LOG(debug, "Message expired from queue '" << name << "'");
- c->position = msg.position;
- acquire( msg.position, msg, locker);
- dequeue( 0, msg );
- continue;
- }
-
- // a message is available for this consumer - can the consumer use it?
+ if (allocator->nextConsumableMessage(c, msg)) {
+ if (msg.payload->hasExpired()) {
+ QPID_LOG(debug, "Message expired from queue '" << name << "'");
+ c->position = msg.position;
+ dequeue(0, msg);
+ continue;
+ }
- if (c->filter(msg.payload)) {
- if (c->accept(msg.payload)) {
- bool ok = allocator->allocate( c->getName(), msg ); // inform allocator
- (void) ok; assert(ok);
- ok = acquire( msg.position, msg, locker);
- (void) ok; assert(ok);
- m = msg;
- c->position = m.position;
- return CONSUMED;
+ if (c->filter(msg.payload)) {
+ if (c->accept(msg.payload)) {
+ bool ok = allocator->allocate( c->getName(), msg ); // inform allocator
+ (void) ok; assert(ok);
+ observeAcquire(msg, locker);
+ m = msg;
+ return CONSUMED;
+ } else {
+ //message(s) are available but consumer hasn't got enough credit
+ QPID_LOG(debug, "Consumer can't currently accept message from '" << name << "'");
+ messages->release(msg);
+ return CANT_CONSUME;
+ }
} else {
- //message(s) are available but consumer hasn't got enough credit
- QPID_LOG(debug, "Consumer can't currently accept message from '" << name << "'");
+ //consumer will never want this message
+ QPID_LOG(debug, "Consumer doesn't want message from '" << name << "'");
+ messages->release(msg);
return CANT_CONSUME;
}
} else {
- //consumer will never want this message
- QPID_LOG(debug, "Consumer doesn't want message from '" << name << "'");
- c->position = msg.position;
- return CANT_CONSUME;
+ QPID_LOG(debug, "No messages to dispatch on queue '" << name << "'");
+ listeners.addListener(c);
+ return NO_MESSAGES;
}
}
}
@@ -398,7 +392,6 @@ bool Queue::dispatch(Consumer::shared_pt
}
bool Queue::find(SequenceNumber pos, QueuedMessage& msg) const {
-
Mutex::ScopedLock locker(messageLock);
if (messages->find(pos, msg))
return true;
@@ -460,7 +453,7 @@ void Queue::cancel(Consumer::shared_ptr
QueuedMessage Queue::get(){
Mutex::ScopedLock locker(messageLock);
QueuedMessage msg(this);
- if (messages->pop(msg))
+ if (messages->consume(msg))
observeAcquire(msg, locker);
return msg;
}
@@ -632,6 +625,7 @@ uint32_t Queue::purge(const uint32_t pur
// Update observers and message state:
observeAcquire(*qmsg, locker);
dequeue(0, *qmsg);
+ QPID_LOG(debug, "Purged message at " << qmsg->position << " from " << getName());
// now reroute if necessary
if (dest.get()) {
assert(qmsg->payload);
@@ -663,24 +657,11 @@ uint32_t Queue::move(const Queue::shared
return c.matches.size();
}
-/** Acquire the front (oldest) message from the in-memory queue.
- * assumes messageLock held by caller
- */
-void Queue::pop(const Mutex::ScopedLock& locker)
-{
- assertClusterSafe();
- QueuedMessage msg;
- if (messages->pop(msg)) {
- observeAcquire(msg, locker);
- ++dequeueSincePurge;
- }
-}
-
/** Acquire the message at the given position, return true and msg if acquire succeeds */
bool Queue::acquire(const qpid::framing::SequenceNumber& position, QueuedMessage& msg,
const Mutex::ScopedLock& locker)
{
- if (messages->remove(position, msg)) {
+ if (messages->acquire(position, msg)) {
observeAcquire(msg, locker);
++dequeueSincePurge;
return true;
@@ -867,12 +848,13 @@ void Queue::dequeueCommitted(const Queue
* Removes the first (oldest) message from the in-memory delivery queue as well dequeing
* it from the logical (and persistent if applicable) queue
*/
-void Queue::popAndDequeue(const Mutex::ScopedLock& held)
+bool Queue::popAndDequeue(QueuedMessage& msg)
{
- if (!messages->empty()) {
- QueuedMessage msg = messages->front();
- pop(held);
+ if (messages->consume(msg)) {
dequeue(0, msg);
+ return true;
+ } else {
+ return false;
}
}
@@ -884,6 +866,7 @@ void Queue::observeDequeue(const QueuedM
{
if (policy.get()) policy->dequeued(msg);
mgntDeqStats(msg.payload);
+ messages->deleted(msg);
for (Observers::const_iterator i = observers.begin(); i != observers.end(); ++i) {
try{
(*i)->dequeued(msg);
@@ -1070,10 +1053,10 @@ void Queue::destroyed()
unbind(broker->getExchanges());
if (alternateExchange.get()) {
Mutex::ScopedLock locker(messageLock);
- while(!messages->empty()){
- DeliverableMessage msg(messages->front().payload);
+ QueuedMessage m;
+ while(popAndDequeue(m)){
+ DeliverableMessage msg(m.payload);
alternateExchange->routeWithAlternate(msg);
- popAndDequeue(locker);
}
alternateExchange->decAlternateUsers();
}
@@ -1336,6 +1319,7 @@ void Queue::query(qpid::types::Variant::
void Queue::setPosition(SequenceNumber n) {
Mutex::ScopedLock locker(messageLock);
sequence = n;
+ QPID_LOG(info, "Set position to " << sequence << " on " << getName());
}
SequenceNumber Queue::getPosition() {
@@ -1422,6 +1406,12 @@ void Queue::addObserver(boost::shared_pt
observers.insert(observer);
}
+void Queue::removeObserver(boost::shared_ptr<QueueObserver> observer)
+{
+ Mutex::ScopedLock locker(messageLock);
+ observers.erase(observer);
+}
+
void Queue::flush()
{
ScopedUse u(barrier);
@@ -1453,6 +1443,38 @@ void Queue::setDequeueSincePurge(uint32_
dequeueSincePurge = value;
}
+namespace{
+class FindLowest
+{
+ public:
+ FindLowest() : init(false) {}
+ void process(const QueuedMessage& message) {
+ QPID_LOG(debug, "FindLowest processing: " << message.position);
+ if (!init || message.position < lowest) lowest = message.position;
+ init = true;
+ }
+ bool getLowest(qpid::framing::SequenceNumber& result) {
+ if (init) {
+ result = lowest;
+ return true;
+ } else {
+ return false;
+ }
+ }
+ private:
+ bool init;
+ qpid::framing::SequenceNumber lowest;
+};
+}
+
+bool Queue::getOldest(qpid::framing::SequenceNumber& oldest)
+{
+ //Horribly inefficient, but saves modifying Messages interface and
+ //all its implementations at present:
+ FindLowest f;
+ eachMessage(boost::bind(&FindLowest::process, &f, _1));
+ return f.getLowest(oldest);
+}
Queue::UsageBarrier::UsageBarrier(Queue& q) : parent(q), count(0) {}
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Queue.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Queue.h?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/Queue.h Wed Nov 9 21:31:07 2011
@@ -148,10 +148,7 @@ class Queue : public boost::enable_share
void observeAcquire(const QueuedMessage& msg, const sys::Mutex::ScopedLock& lock);
void observeRequeue(const QueuedMessage& msg, const sys::Mutex::ScopedLock& lock);
void observeDequeue(const QueuedMessage& msg, const sys::Mutex::ScopedLock& lock);
-
- /** modify the Queue's message container - assumes messageLock held */
- void pop(const sys::Mutex::ScopedLock& held); // acquire front msg
- void popAndDequeue(const sys::Mutex::ScopedLock& held); // acquire and dequeue front msg
+ bool popAndDequeue(QueuedMessage&);
// acquire message @ position, return true and set msg if acquire succeeds
bool acquire(const qpid::framing::SequenceNumber& position, QueuedMessage& msg,
const sys::Mutex::ScopedLock& held);
@@ -386,6 +383,7 @@ class Queue : public boost::enable_share
*/
QPID_BROKER_EXTERN framing::SequenceNumber getPosition();
void addObserver(boost::shared_ptr<QueueObserver>);
+ void removeObserver(boost::shared_ptr<QueueObserver>);
QPID_BROKER_EXTERN void insertSequenceNumbers(const std::string& key);
/**
* Notify queue that recovery has completed.
@@ -409,6 +407,7 @@ class Queue : public boost::enable_share
uint32_t getDequeueSincePurge() { return dequeueSincePurge.get(); }
void setDequeueSincePurge(uint32_t value);
+ bool getOldest(framing::SequenceNumber& result);
};
}
}
Added: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.cpp?rev=1199975&view=auto
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.cpp (added)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.cpp Wed Nov 9 21:31:07 2011
@@ -0,0 +1,128 @@
+/*
+ *
+ * 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 "qpid/broker/QueueReplicator.h"
+#include "qpid/broker/Queue.h"
+#include "qpid/broker/QueueRegistry.h"
+#include "qpid/framing/SequenceSet.h"
+#include "qpid/log/Statement.h"
+
+namespace qpid {
+namespace broker {
+
+QueueReplicator::QueueReplicator(const std::string& name, boost::shared_ptr<Queue> q) : Exchange(name, 0, 0), queue(q), current(queue->getPosition()) {}
+QueueReplicator::~QueueReplicator() {}
+
+namespace {
+const std::string DEQUEUE_EVENT("dequeue-event");
+const std::string REPLICATOR("qpid.replicator-");
+}
+
+void QueueReplicator::route(Deliverable& msg, const std::string& key, const qpid::framing::FieldTable* /*args*/)
+{
+ if (key == DEQUEUE_EVENT) {
+ std::string content;
+ msg.getMessage().getFrames().getContent(content);
+ qpid::framing::Buffer buffer(const_cast<char*>(content.c_str()), content.size());
+ qpid::framing::SequenceSet latest;
+ latest.decode(buffer);
+
+ //TODO: should be able to optimise the following
+ for (qpid::framing::SequenceSet::iterator i = latest.begin(); i != latest.end(); i++) {
+ if (current < *i) {
+ //haven't got that far yet, record the dequeue
+ dequeued.add(*i);
+ QPID_LOG(debug, "Recording dequeue of message at " << *i << " from " << queue->getName());
+ } else {
+ QueuedMessage message;
+ if (queue->acquireMessageAt(*i, message)) {
+ queue->dequeue(0, message);
+ QPID_LOG(info, "Dequeued message at " << *i << " from " << queue->getName());
+ } else {
+ QPID_LOG(error, "Unable to dequeue message at " << *i << " from " << queue->getName());
+ }
+ }
+ }
+ } else {
+ //take account of any gaps in sequence created by messages
+ //dequeued before our subscription reached them
+ while (dequeued.contains(++current)) {
+ dequeued.remove(current);
+ QPID_LOG(debug, "Skipping dequeued message at " << current << " from " << queue->getName());
+ queue->setPosition(current);
+ }
+ QPID_LOG(info, "Enqueued message on " << queue->getName() << "; currently at " << current);
+ msg.deliverTo(queue);
+ }
+}
+
+bool QueueReplicator::isReplicatingLink(const std::string& name)
+{
+ return name.find(REPLICATOR) == 0;
+}
+
+boost::shared_ptr<Exchange> QueueReplicator::create(const std::string& target, QueueRegistry& queues)
+{
+ boost::shared_ptr<Exchange> exchange;
+ if (isReplicatingLink(target)) {
+ std::string queueName = target.substr(REPLICATOR.size());
+ boost::shared_ptr<Queue> queue = queues.find(queueName);
+ if (!queue) {
+ QPID_LOG(warning, "Unable to create replicator, can't find " << queueName);
+ } else {
+ //TODO: need to cache the replicator
+ QPID_LOG(info, "Creating replicator for " << queueName);
+ exchange.reset(new QueueReplicator(target, queue));
+ }
+ }
+ return exchange;
+}
+
+bool QueueReplicator::initReplicationSettings(const std::string& target, QueueRegistry& queues, qpid::framing::FieldTable& settings)
+{
+ if (isReplicatingLink(target)) {
+ std::string queueName = target.substr(REPLICATOR.size());
+ boost::shared_ptr<Queue> queue = queues.find(queueName);
+ if (queue) {
+ settings.setInt("qpid.replicating-subscription", 1);
+ settings.setInt("qpid.high_sequence_number", queue->getPosition());
+ qpid::framing::SequenceNumber oldest;
+ if (queue->getOldest(oldest)) {
+ settings.setInt("qpid.low_sequence_number", oldest);
+ }
+ }
+ return true;
+ } else {
+ return false;
+ }
+}
+
+bool QueueReplicator::bind(boost::shared_ptr<Queue>, const std::string&, const qpid::framing::FieldTable*) { return false; }
+bool QueueReplicator::unbind(boost::shared_ptr<Queue>, const std::string&, const qpid::framing::FieldTable*) { return false; }
+bool QueueReplicator::isBound(boost::shared_ptr<Queue>, const std::string* const, const qpid::framing::FieldTable* const) { return false; }
+
+const std::string QueueReplicator::typeName("queue-replicator");
+
+std::string QueueReplicator::getType() const
+{
+ return typeName;
+}
+
+}} // namespace qpid::broker
Propchange: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.h?rev=1199975&view=auto
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.h (added)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.h Wed Nov 9 21:31:07 2011
@@ -0,0 +1,57 @@
+#ifndef QPID_BROKER_QUEUEREPLICATOR_H
+#define QPID_BROKER_QUEUEREPLICATOR_H
+
+/*
+ *
+ * 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 "qpid/broker/Exchange.h"
+#include "qpid/framing/SequenceSet.h"
+
+namespace qpid {
+namespace broker {
+
+class QueueRegistry;
+
+/**
+ * Dummy exchange for processing replication messages
+ */
+class QueueReplicator : public Exchange
+{
+ public:
+ QueueReplicator(const std::string& name, boost::shared_ptr<Queue>);
+ ~QueueReplicator();
+ std::string getType() const;
+ bool bind(boost::shared_ptr<Queue>, const std::string&, const qpid::framing::FieldTable*);
+ bool unbind(boost::shared_ptr<Queue>, const std::string&, const qpid::framing::FieldTable*);
+ void route(Deliverable&, const std::string&, const qpid::framing::FieldTable*);
+ bool isBound(boost::shared_ptr<Queue>, const std::string* const, const qpid::framing::FieldTable* const);
+ static bool isReplicatingLink(const std::string&);
+ static boost::shared_ptr<Exchange> create(const std::string&, QueueRegistry&);
+ static bool initReplicationSettings(const std::string&, QueueRegistry&, qpid::framing::FieldTable&);
+ static const std::string typeName;
+ private:
+ boost::shared_ptr<Queue> queue;
+ qpid::framing::SequenceNumber current;
+ qpid::framing::SequenceSet dequeued;
+};
+
+}} // namespace qpid::broker
+
+#endif /*!QPID_BROKER_QUEUEREPLICATOR_H*/
Propchange: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueueReplicator.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueuedMessage.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueuedMessage.h?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueuedMessage.h (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/QueuedMessage.h Wed Nov 9 21:31:07 2011
@@ -32,6 +32,7 @@ struct QueuedMessage
{
boost::intrusive_ptr<Message> payload;
framing::SequenceNumber position;
+ enum {AVAILABLE, ACQUIRED, DELETED, REMOVED} status;
Queue* queue;
QueuedMessage() : queue(0) {}
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/SemanticState.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/SemanticState.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/SemanticState.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/SemanticState.cpp Wed Nov 9 21:31:07 2011
@@ -26,6 +26,7 @@
#include "qpid/broker/DtxTimeout.h"
#include "qpid/broker/Message.h"
#include "qpid/broker/Queue.h"
+#include "qpid/broker/QueueReplicator.h"
#include "qpid/broker/SessionContext.h"
#include "qpid/broker/SessionOutputException.h"
#include "qpid/broker/TxAccept.h"
@@ -116,7 +117,7 @@ void SemanticState::consume(const string
// "tag" is only guaranteed to be unique to this session (see AMQP 0-10 Message.subscribe, destination).
// Create a globally unique name so the broker can identify individual consumers
std::string name = session.getSessionId().str() + SEPARATOR + tag;
- ConsumerImpl::shared_ptr c(new ConsumerImpl(this, name, queue, ackRequired, acquire, exclusive, tag, resumeId, resumeTtl, arguments));
+ ConsumerImpl::shared_ptr c(ConsumerImpl::create(this, name, queue, ackRequired, acquire, exclusive, tag, resumeId, resumeTtl, arguments));
queue->consume(c, exclusive);//may throw exception
consumers[tag] = c;
}
@@ -266,6 +267,223 @@ void SemanticState::record(const Deliver
const std::string QPID_SYNC_FREQUENCY("qpid.sync_frequency");
+class ReplicatingSubscription : public SemanticState::ConsumerImpl, public QueueObserver
+{
+ public:
+ ReplicatingSubscription(SemanticState* parent,
+ const std::string& name, boost::shared_ptr<Queue> queue,
+ bool ack, bool acquire, bool exclusive, const std::string& tag,
+ const std::string& resumeId, uint64_t resumeTtl, const framing::FieldTable& arguments);
+ ~ReplicatingSubscription();
+
+ void init();
+ void cancel();
+ bool deliver(QueuedMessage& msg);
+ void enqueued(const QueuedMessage&);
+ void dequeued(const QueuedMessage&);
+ void acquired(const QueuedMessage&) {}
+ void requeued(const QueuedMessage&) {}
+
+ protected:
+ bool doDispatch();
+ private:
+ boost::shared_ptr<Queue> events;
+ boost::shared_ptr<Consumer> consumer;
+ qpid::framing::SequenceSet range;
+
+ void generateDequeueEvent();
+ class DelegatingConsumer : public Consumer
+ {
+ public:
+ DelegatingConsumer(ReplicatingSubscription&);
+ ~DelegatingConsumer();
+ bool deliver(QueuedMessage& msg);
+ void notify();
+ bool filter(boost::intrusive_ptr<Message>);
+ bool accept(boost::intrusive_ptr<Message>);
+ OwnershipToken* getSession();
+ private:
+ ReplicatingSubscription& delegate;
+ };
+};
+
+SemanticState::ConsumerImpl::shared_ptr SemanticState::ConsumerImpl::create(SemanticState* parent,
+ const string& name,
+ Queue::shared_ptr queue,
+ bool ack,
+ bool acquire,
+ bool exclusive,
+ const string& tag,
+ const string& resumeId,
+ uint64_t resumeTtl,
+ const framing::FieldTable& arguments)
+{
+ if (arguments.isSet("qpid.replicating-subscription")) {
+ shared_ptr result(new ReplicatingSubscription(parent, name, queue, ack, acquire, exclusive, tag, resumeId, resumeTtl, arguments));
+ boost::dynamic_pointer_cast<ReplicatingSubscription>(result)->init();
+ return result;
+ } else {
+ return shared_ptr(new ConsumerImpl(parent, name, queue, ack, acquire, exclusive, tag, resumeId, resumeTtl, arguments));
+ }
+}
+
+std::string mask(const std::string& in)
+{
+ return std::string("$") + in + std::string("_internal");
+}
+
+class ReplicationStateInitialiser
+{
+ public:
+ ReplicationStateInitialiser(qpid::framing::SequenceSet& results,
+ const qpid::framing::SequenceNumber& start,
+ const qpid::framing::SequenceNumber& end);
+ void operator()(const QueuedMessage& m) { process(m); }
+ private:
+ qpid::framing::SequenceSet& results;
+ const qpid::framing::SequenceNumber start;
+ const qpid::framing::SequenceNumber end;
+ void process(const QueuedMessage&);
+};
+
+ReplicatingSubscription::ReplicatingSubscription(SemanticState* _parent,
+ const string& _name,
+ Queue::shared_ptr _queue,
+ bool ack,
+ bool _acquire,
+ bool _exclusive,
+ const string& _tag,
+ const string& _resumeId,
+ uint64_t _resumeTtl,
+ const framing::FieldTable& _arguments
+) : ConsumerImpl(_parent, _name, _queue, ack, _acquire, _exclusive, _tag, _resumeId, _resumeTtl, _arguments),
+ events(new Queue(mask(_name))),
+ consumer(new DelegatingConsumer(*this))
+{
+
+ if (_arguments.isSet("qpid.high_sequence_number")) {
+ qpid::framing::SequenceNumber hwm = _arguments.getAsInt("qpid.high_sequence_number");
+ qpid::framing::SequenceNumber lwm;
+ if (_arguments.isSet("qpid.low_sequence_number")) {
+ lwm = _arguments.getAsInt("qpid.low_sequence_number");
+ } else {
+ lwm = hwm;
+ }
+ qpid::framing::SequenceNumber oldest;
+ if (_queue->getOldest(oldest)) {
+ if (oldest >= hwm) {
+ range.add(lwm, --oldest);
+ } else if (oldest >= lwm) {
+ ReplicationStateInitialiser initialiser(range, lwm, hwm);
+ _queue->eachMessage(initialiser);
+ } else { //i.e. have older message on master than is reported to exist on replica
+ QPID_LOG(warning, "Replica appears to be missing message on master");
+ }
+ } else {
+ //local queue (i.e. master) is empty
+ range.add(lwm, _queue->getPosition());
+ }
+ QPID_LOG(debug, "Initial set of dequeues for " << _queue->getName() << " are " << range
+ << " (lwm=" << lwm << ", hwm=" << hwm << ", current=" << _queue->getPosition() << ")");
+ //set position of 'cursor'
+ position = hwm;
+ }
+}
+
+bool ReplicatingSubscription::deliver(QueuedMessage& m)
+{
+ return ConsumerImpl::deliver(m);
+}
+
+void ReplicatingSubscription::init()
+{
+ getQueue()->addObserver(boost::dynamic_pointer_cast<QueueObserver>(shared_from_this()));
+}
+
+void ReplicatingSubscription::cancel()
+{
+ getQueue()->removeObserver(boost::dynamic_pointer_cast<QueueObserver>(shared_from_this()));
+}
+
+ReplicatingSubscription::~ReplicatingSubscription() {}
+
+//called before we get notified of the message being available and
+//under the message lock in the queue
+void ReplicatingSubscription::enqueued(const QueuedMessage& m)
+{
+ QPID_LOG(debug, "Enqueued message at " << m.position);
+ //delay completion
+ m.payload->getIngressCompletion().startCompleter();
+ QPID_LOG(debug, "Delayed " << m.payload.get());
+}
+
+class Buffer : public qpid::framing::Buffer
+{
+ public:
+ Buffer(size_t size) : qpid::framing::Buffer(new char[size], size) {}
+ ~Buffer() { delete[] getPointer(); }
+};
+
+void ReplicatingSubscription::generateDequeueEvent()
+{
+ Buffer buffer(range.encodedSize());
+ range.encode(buffer);
+ range.clear();
+ buffer.reset();
+
+ //generate event message
+ boost::intrusive_ptr<Message> event = new Message();
+ AMQFrame method((MessageTransferBody(ProtocolVersion(), std::string(), 0, 0)));
+ AMQFrame header((AMQHeaderBody()));
+ AMQFrame content((AMQContentBody()));
+ content.castBody<AMQContentBody>()->decode(buffer, buffer.getSize());
+ header.setBof(false);
+ header.setEof(false);
+ header.setBos(true);
+ header.setEos(true);
+ content.setBof(false);
+ content.setEof(true);
+ content.setBos(true);
+ content.setEos(true);
+ event->getFrames().append(method);
+ event->getFrames().append(header);
+ event->getFrames().append(content);
+
+ DeliveryProperties* props = event->getFrames().getHeaders()->get<DeliveryProperties>(true);
+ props->setRoutingKey("dequeue-event");
+
+ events->deliver(event);
+}
+
+//called after the message has been removed from the deque and under
+//the message lock in the queue
+void ReplicatingSubscription::dequeued(const QueuedMessage& m)
+{
+ {
+ Mutex::ScopedLock l(lock);
+ range.add(m.position);
+ QPID_LOG(debug, "Updated dequeue event to include message at " << m.position << "; subscription is at " << position);
+ }
+ notify();
+ if (m.position > position) {
+ m.payload->getIngressCompletion().finishCompleter();
+ QPID_LOG(debug, "Completed " << m.payload.get() << " early due to dequeue");
+ }
+}
+
+bool ReplicatingSubscription::doDispatch()
+{
+ {
+ Mutex::ScopedLock l(lock);
+ if (!range.empty()) {
+ generateDequeueEvent();
+ }
+ }
+ bool r1 = events->dispatch(consumer);
+ bool r2 = ConsumerImpl::doDispatch();
+ return r1 || r2;
+}
+
SemanticState::ConsumerImpl::ConsumerImpl(SemanticState* _parent,
const string& _name,
Queue::shared_ptr _queue,
@@ -338,7 +556,7 @@ bool SemanticState::ConsumerImpl::delive
{
assertClusterSafe();
allocateCredit(msg.payload);
- DeliveryRecord record(msg, queue, getTag(), acquire, !ackExpected, windowing);
+ DeliveryRecord record(msg, msg.queue->shared_from_this(), getTag(), acquire, !ackExpected, windowing, 0, dynamic_cast<const ReplicatingSubscription*>(this));
bool sync = syncFrequency && ++deliveryCount >= syncFrequency;
if (sync) deliveryCount = 0;//reset
parent->deliver(record, sync);
@@ -346,7 +564,7 @@ bool SemanticState::ConsumerImpl::delive
parent->record(record);
}
if (acquire && !ackExpected) { // auto acquire && auto accept
- queue->dequeue(0 /*ctxt*/, msg);
+ msg.queue->dequeue(0, msg);
record.setEnded();
}
if (mgmtObject) { mgmtObject->inc_delivered(); }
@@ -436,6 +654,7 @@ void SemanticState::unsubscribe(Consumer
void SemanticState::cancel(ConsumerImpl::shared_ptr c)
{
+ c->cancel();
disable(c);
unsubscribe(c);
}
@@ -473,8 +692,10 @@ void SemanticState::route(intrusive_ptr<
msg->computeExpiration(getSession().getBroker().getExpiryPolicy());
std::string exchangeName = msg->getExchangeName();
- if (!cacheExchange || cacheExchange->getName() != exchangeName || cacheExchange->isDestroyed())
- cacheExchange = session.getBroker().getExchanges().get(exchangeName);
+ if (!cacheExchange || cacheExchange->getName() != exchangeName || cacheExchange->isDestroyed()) {
+ cacheExchange = QueueReplicator::create(exchangeName, getSession().getBroker().getQueues());
+ if (!cacheExchange) cacheExchange = session.getBroker().getExchanges().get(exchangeName);
+ }
cacheExchange->setProperties(msg);
/* verify the userid if specified: */
@@ -673,9 +894,14 @@ bool SemanticState::ConsumerImpl::haveCr
}
}
+bool SemanticState::ConsumerImpl::doDispatch()
+{
+ return queue->dispatch(shared_from_this());
+}
+
void SemanticState::ConsumerImpl::flush()
{
- while(haveCredit() && queue->dispatch(shared_from_this()))
+ while(haveCredit() && doDispatch())
;
msgCredit = 0;
byteCredit = 0;
@@ -740,7 +966,7 @@ void SemanticState::reject(DeliveryId fi
bool SemanticState::ConsumerImpl::doOutput()
{
try {
- return haveCredit() && queue->dispatch(shared_from_this());
+ return haveCredit() && doDispatch();
} catch (const SessionException& e) {
throw SessionOutputException(e, parent->session.getChannel());
}
@@ -850,4 +1076,35 @@ void SemanticState::detached()
}
}
+ReplicatingSubscription::DelegatingConsumer::DelegatingConsumer(ReplicatingSubscription& c) : Consumer(c.getName(), true), delegate(c) {}
+ReplicatingSubscription::DelegatingConsumer::~DelegatingConsumer() {}
+bool ReplicatingSubscription::DelegatingConsumer::deliver(QueuedMessage& m)
+{
+ return delegate.deliver(m);
+}
+void ReplicatingSubscription::DelegatingConsumer::notify() { delegate.notify(); }
+bool ReplicatingSubscription::DelegatingConsumer::filter(boost::intrusive_ptr<Message> msg) { return delegate.filter(msg); }
+bool ReplicatingSubscription::DelegatingConsumer::accept(boost::intrusive_ptr<Message> msg) { return delegate.accept(msg); }
+OwnershipToken* ReplicatingSubscription::DelegatingConsumer::getSession() { return delegate.getSession(); }
+
+ReplicationStateInitialiser::ReplicationStateInitialiser(qpid::framing::SequenceSet& r,
+ const qpid::framing::SequenceNumber& s,
+ const qpid::framing::SequenceNumber& e)
+ : results(r), start(s), end(e)
+{
+ results.add(start, end);
+}
+
+void ReplicationStateInitialiser::process(const QueuedMessage& message)
+{
+ if (message.position < start) {
+ //replica does not have a message that should still be on the queue
+ QPID_LOG(warning, "Replica appears to be missing message at " << message.position);
+ } else if (message.position >= start && message.position <= end) {
+ //i.e. message is within the intial range and has not been dequeued, so remove it from the results
+ results.remove(message.position);
+ } //else message has not been seen by replica yet so can be ignored here
+
+}
+
}} // namespace qpid::broker
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/SemanticState.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/SemanticState.h?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/SemanticState.h (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/broker/SemanticState.h Wed Nov 9 21:31:07 2011
@@ -29,6 +29,7 @@
#include "qpid/broker/DtxBuffer.h"
#include "qpid/broker/DtxManager.h"
#include "qpid/broker/NameGenerator.h"
+#include "qpid/broker/QueueObserver.h"
#include "qpid/broker/TxBuffer.h"
#include "qpid/framing/FrameHandler.h"
@@ -73,7 +74,9 @@ class SemanticState : private boost::non
public boost::enable_shared_from_this<ConsumerImpl>,
public management::Manageable
{
+ protected:
mutable qpid::sys::Mutex lock;
+ private:
SemanticState* const parent;
const boost::shared_ptr<Queue> queue;
const bool ackExpected;
@@ -97,17 +100,20 @@ class SemanticState : private boost::non
void allocateCredit(boost::intrusive_ptr<Message>& msg);
bool haveCredit();
+ protected:
+ virtual bool doDispatch();
+ size_t unacked() { return parent->unacked.size(); }
+
public:
typedef boost::shared_ptr<ConsumerImpl> shared_ptr;
ConsumerImpl(SemanticState* parent,
const std::string& name, boost::shared_ptr<Queue> queue,
bool ack, bool acquire, bool exclusive,
- const std::string& tag, const std::string& resumeId,
- uint64_t resumeTtl, const framing::FieldTable& arguments);
- ~ConsumerImpl();
+ const std::string& tag, const std::string& resumeId, uint64_t resumeTtl, const framing::FieldTable& arguments);
+ virtual ~ConsumerImpl();
OwnershipToken* getSession();
- bool deliver(QueuedMessage& msg);
+ virtual bool deliver(QueuedMessage& msg);
bool filter(boost::intrusive_ptr<Message> msg);
bool accept(boost::intrusive_ptr<Message> msg);
@@ -144,9 +150,16 @@ class SemanticState : private boost::non
SemanticState& getParent() { return *parent; }
const SemanticState& getParent() const { return *parent; }
+ virtual void cancel() {}
// Manageable entry points
management::ManagementObject* GetManagementObject (void) const;
management::Manageable::status_t ManagementMethod (uint32_t methodId, management::Args& args, std::string& text);
+
+ static shared_ptr create(SemanticState* parent,
+ const std::string& name, boost::shared_ptr<Queue> queue,
+ bool ack, bool acquire, bool exclusive, const std::string& tag,
+ const std::string& resumeId, uint64_t resumeTtl, const framing::FieldTable& arguments);
+
};
typedef std::map<std::string, DtxBuffer::shared_ptr> DtxBufferMap;
Modified: qpid/branches/qpid-3603/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp Wed Nov 9 21:31:07 2011
@@ -275,7 +275,7 @@ void IncomingMessages::retrieve(FrameSet
populate(*message, *command);
}
const MessageTransferBody* transfer = command->as<MessageTransferBody>();
- if (transfer->getAcquireMode() == ACQUIRE_MODE_PRE_ACQUIRED && transfer->getAcceptMode() == ACCEPT_MODE_EXPLICIT) {
+ if (transfer->getAcceptMode() == ACCEPT_MODE_EXPLICIT) {
sys::Mutex::ScopedLock l(lock);
acceptTracker.delivered(transfer->getDestination(), command->getId());
}
Modified: qpid/branches/qpid-3603/qpid/cpp/src/tests/QueueTest.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/cpp/src/tests/QueueTest.cpp?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/cpp/src/tests/QueueTest.cpp (original)
+++ qpid/branches/qpid-3603/qpid/cpp/src/tests/QueueTest.cpp Wed Nov 9 21:31:07 2011
@@ -710,7 +710,7 @@ namespace {
const std::string& expectedGroup,
const int expectedId )
{
- queue->dispatch(c);
+ BOOST_CHECK(queue->dispatch(c));
results.push_back(c->last);
std::string group = c->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsString("GROUP-ID");
int id = c->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsInt("MY-ID");
@@ -1025,6 +1025,11 @@ QPID_AUTO_TEST_CASE(testMultiQueueLastNo
queue2->setLastNodeFailure();
BOOST_CHECK_EQUAL(testStore.enqCnt, 6u);
+ /**
+ * TODO: Fix or replace the following test which incorrectly requeues a
+ * message that was never on the queue in the first place. This relied on
+ * internal details not part of the queue abstraction.
+
// check requeue 1
intrusive_ptr<Message> msg4 = create_message("e", "C");
intrusive_ptr<Message> msg5 = create_message("e", "D");
@@ -1046,6 +1051,7 @@ QPID_AUTO_TEST_CASE(testMultiQueueLastNo
queue2->clearLastNodeFailure();
queue2->setLastNodeFailure();
BOOST_CHECK_EQUAL(testStore.enqCnt, 8u);
+ */
}
QPID_AUTO_TEST_CASE(testLastNodeRecoverAndFail){
Modified: qpid/branches/qpid-3603/qpid/python/qpid/tests/messaging/endpoints.py
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/python/qpid/tests/messaging/endpoints.py?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/python/qpid/tests/messaging/endpoints.py (original)
+++ qpid/branches/qpid-3603/qpid/python/qpid/tests/messaging/endpoints.py Wed Nov 9 21:31:07 2011
@@ -886,9 +886,11 @@ class ReceiverTests(Base):
rc = self.ssn.receiver('test-receiver-queue; {mode: consume}')
self.drain(rb, expected=msgs)
self.drain(rc, expected=msgs)
- rb2 = self.ssn.receiver(rb.source)
- self.assertEmpty(rb2)
+ rc2 = self.ssn.receiver(rc.source)
+ self.assertEmpty(rc2)
self.drain(self.rcv, expected=[])
+ rb2 = self.ssn.receiver(rb.source)
+ self.drain(rb2, expected=msgs)
# XXX: need testUnsettled()
Modified: qpid/branches/qpid-3603/qpid/tests/src/py/qpid_tests/broker_0_10/message.py
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/tests/src/py/qpid_tests/broker_0_10/message.py?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/tests/src/py/qpid_tests/broker_0_10/message.py (original)
+++ qpid/branches/qpid-3603/qpid/tests/src/py/qpid_tests/broker_0_10/message.py Wed Nov 9 21:31:07 2011
@@ -955,8 +955,7 @@ class MessageTests(TestBase010):
#release all even messages
session.message_release(RangedSet(msg.id))
- #browse:
- session.message_subscribe(queue="q", destination="b", acquire_mode=1)
+ session.message_subscribe(queue="q", destination="b", acquire_mode=0)
b = session.incoming("b")
b.start()
for i in [2, 4, 6, 8, 10]:
Modified: qpid/branches/qpid-3603/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py?rev=1199975&r1=1199974&r2=1199975&view=diff
==============================================================================
--- qpid/branches/qpid-3603/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py (original)
+++ qpid/branches/qpid-3603/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py Wed Nov 9 21:31:07 2011
@@ -202,6 +202,10 @@ class MultiConsumerMsgGroupTests(Base):
## Queue = A-0, B-1, A-2, b-3, C-4
## Owners= ^C1, ---, +C1, ---, ---
+ m2 = b1.fetch(0);
+ assert m2.properties['THE-GROUP'] == 'A'
+ assert m2.content['index'] == 0
+
m2 = b1.fetch(0)
assert m2.properties['THE-GROUP'] == 'B'
assert m2.content['index'] == 1
@@ -713,6 +717,7 @@ class MultiConsumerMsgGroupTests(Base):
assert rc.status == 0
queue.update()
queue.msgDepth == 4 # the pending acquired A still counts!
+ s1.acknowledge()
# verify all other A's removed....
s2 = self.setup_session()
@@ -782,7 +787,7 @@ class MultiConsumerMsgGroupTests(Base):
except Empty:
pass
assert count == 3 # non-A's
- assert a_count == 1 # and one is an A
+ assert a_count == 2 # pending acquired message included in browse results
s1.acknowledge() # ack the consumed A-0
self.qmf_session.delBroker(self.qmf_broker)
@@ -829,7 +834,7 @@ class MultiConsumerMsgGroupTests(Base):
# verify all other A's removed from msg-group-q
s2 = self.setup_session()
- b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":0})
+ b1 = s2.receiver("msg-group-q", options={"capacity":0})
count = 0
try:
while True:
@@ -963,7 +968,7 @@ class MultiConsumerMsgGroupTests(Base):
# verify all other A's removed....
s2 = self.setup_session()
- b1 = s2.receiver("msg-group-q; {mode: browse}", options={"capacity":0})
+ b1 = s2.receiver("msg-group-q", options={"capacity":0})
count = 0
try:
while True:
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org