You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2009/02/13 20:21:21 UTC
svn commit: r744222 - in /qpid/trunk/qpid/cpp/src: qpid/broker/Queue.cpp
tests/ClientSessionTest.cpp
Author: gsim
Date: Fri Feb 13 19:21:21 2009
New Revision: 744222
URL: http://svn.apache.org/viewvc?rev=744222&view=rev
Log:
Ensure that the queue depth in bytes remains accurate for LVQ. This also ensures that there are no underflow exceptions from the policy that keeps this count.
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=744222&r1=744221&r2=744222&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Fri Feb 13 19:21:21 2009
@@ -563,8 +563,10 @@
listeners.populate(copy);
lvq[key] = msg;
}else {
+ boost::intrusive_ptr<Message> old = i->second->getReplacementMessage(this);
+ if (!old) old = i->second;
i->second->setReplacementMessage(msg,this);
- dequeued(QueuedMessage(qm.queue, i->second, qm.position));
+ dequeued(QueuedMessage(qm.queue, old, qm.position));
}
}else {
messages.push_back(qm);
Modified: qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp?rev=744222&r1=744221&r2=744222&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp Fri Feb 13 19:21:21 2009
@@ -21,6 +21,7 @@
#include "unit_test.h"
#include "test_tools.h"
#include "BrokerFixture.h"
+#include "qpid/client/QueueOptions.h"
#include "qpid/client/SubscriptionManager.h"
#include "qpid/sys/Monitor.h"
#include "qpid/sys/Thread.h"
@@ -526,6 +527,27 @@
session.close();
}
+QPID_AUTO_TEST_CASE(testLVQVariedSize) {
+ ClientSessionFixture fix;
+ std::string queue("my-lvq");
+ QueueOptions args;
+ args.setOrdering(LVQ_NO_BROWSE);
+ fix.session.queueDeclare(arg::queue=queue, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args);
+
+ std::string key;
+ args.getLVQKey(key);
+
+ for (size_t i = 0; i < 10; i++) {
+ std::ostringstream data;
+ size_t size = 100 - ((i % 10) * 10);
+ data << std::string(size, 'x');
+
+ Message m(data.str(), queue);
+ m.getHeaders().setString(key, "abc");
+ fix.session.messageTransfer(arg::content=m);
+ }
+}
+
QPID_AUTO_TEST_SUITE_END()
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org