Author: gsim
Date: Thu Oct 4 09:06:05 2007
New Revision: 581957
URL: http://svn.apache.org/viewvc?rev=581957&view=rev
Log:
Additional tests and fixes
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AccumulatedAck.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AccumulatedAck.h
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h
incubator/qpid/trunk/qpid/cpp/src/tests/AccumulatedAckTest.cpp
incubator/qpid/trunk/qpid/python/tests_0-10/message.py
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp?rev=581957&r1=581956&r2=581957&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp
(original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp Thu
Oct 4 09:06:05 2007
@@ -198,7 +198,8 @@
//TODO: implement mode
SequenceNumberSet results;
- transfers.processRanges(boost::bind(&SemanticState::acquire, &state, _1,
_2, results));
+ RangedOperation op = boost::bind(&SemanticState::acquire, &state, _1, _2,
results);
+ transfers.processRanges(op);
results = results.condense();
getProxy().getMessage().acquired(results);
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp?rev=581957&r1=581956&r2=581957&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp Thu Oct
4 09:06:05 2007
@@ -31,12 +31,16 @@
#include "qpid/framing/InvocationVisitor.h"
#include <boost/format.hpp>
+#include <boost/bind.hpp>
using namespace qpid::broker;
using namespace qpid::framing;
using namespace qpid::sys;
-SemanticHandler::SemanticHandler(SessionState& s) : state(*this,s), session(s)
{}
+SemanticHandler::SemanticHandler(SessionState& s) :
+ state(*this,s), session(s),
+ ackOp(boost::bind(&SemanticState::ackRange, &state, _1, _2))
+ {}
void SemanticHandler::handle(framing::AMQFrame& frame)
{
@@ -81,13 +85,7 @@
//ack messages:
state.ackCumulative(mark.getValue());
}
- if (range.size() % 2) { //must be even number
- throw ConnectionException(530, "Received odd number of elements in
ranged mark");
- } else {
- for (SequenceNumberSet::const_iterator i = range.begin(); i !=
range.end(); i++) {
- state.ackRange(*i, *(++i));
- }
- }
+ range.processRanges(ackOp);
}
void SemanticHandler::sendCompletion()
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h?rev=581957&r1=581956&r2=581957&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h Thu Oct 4
09:06:05 2007
@@ -33,6 +33,8 @@
#include "qpid/framing/FrameHandler.h"
#include "qpid/framing/SequenceNumber.h"
+#include <boost/function.hpp>
+
namespace qpid {
namespace framing {
@@ -51,6 +53,8 @@
public framing::AMQP_ServerOperations::ExecutionHandler
{
+ typedef boost::function<void(DeliveryId, DeliveryId)> RangedOperation;
+
SemanticState state;
SessionState& session;
// FIXME aconway 2007-09-20: Why are these on the handler rather than the
@@ -59,6 +63,7 @@
framing::Window outgoing;
sys::Mutex outLock;
MessageBuilder msgBuilder;
+ RangedOperation ackOp;
enum TrackId {EXECUTION_CONTROL_TRACK, MODEL_COMMAND_TRACK,
MODEL_CONTENT_TRACK};
TrackId getTrack(const framing::AMQFrame& frame);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AccumulatedAck.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AccumulatedAck.cpp?rev=581957&r1=581956&r2=581957&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AccumulatedAck.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AccumulatedAck.cpp Thu Oct
4 09:06:05 2007
@@ -22,12 +22,15 @@
#include <assert.h>
#include <iostream>
+#include <boost/bind.hpp>
using std::list;
using std::max;
using std::min;
using namespace qpid::framing;
+AccumulatedAck::AccumulatedAck(SequenceNumber r) : mark(r) {}
+
void AccumulatedAck::update(SequenceNumber first, SequenceNumber last){
assert(first <= last);
if (last < mark) return;
@@ -103,9 +106,7 @@
void AccumulatedAck::update(const SequenceNumber cumulative, const
SequenceNumberSet& range)
{
update(mark, cumulative);
- for (SequenceNumberSet::const_iterator i = range.begin(); i !=
range.end(); i++) {
- update(*i, *(++i));
- }
+ range.processRanges(*this);
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AccumulatedAck.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AccumulatedAck.h?rev=581957&r1=581956&r2=581957&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AccumulatedAck.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AccumulatedAck.h Thu Oct 4
09:06:05 2007
@@ -58,13 +58,14 @@
*/
std::list<Range> ranges;
- explicit AccumulatedAck(SequenceNumber r = SequenceNumber()) :
mark(r) {}
+ explicit AccumulatedAck(SequenceNumber r = SequenceNumber());
void update(SequenceNumber firstTag, SequenceNumber lastTag);
void consolidate();
void clear();
bool covers(SequenceNumber tag) const;
void collectRanges(SequenceNumberSet& set) const;
void update(const SequenceNumber cumulative, const
SequenceNumberSet& range);
+ void operator()(SequenceNumber first, SequenceNumber last) {
update(first, last); }
};
std::ostream& operator<<(std::ostream&, const Range&);
std::ostream& operator<<(std::ostream&, const AccumulatedAck&);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp?rev=581957&r1=581956&r2=581957&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp
(original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp Thu
Oct 4 09:06:05 2007
@@ -60,6 +60,12 @@
return result;
}
+void SequenceNumberSet::addRange(const SequenceNumber& start, const
SequenceNumber& end)
+{
+ push_back(start);
+ push_back(end);
+}
+
namespace qpid{
namespace framing{
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h?rev=581957&r1=581956&r2=581957&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h
(original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h Thu Oct
4 09:06:05 2007
@@ -41,17 +41,18 @@
void decode(Buffer& buffer);
uint32_t encodedSize() const;
SequenceNumberSet condense() const;
+ void addRange(const SequenceNumber& start, const SequenceNumber& end);
template <class T>
- void processRanges(T t) const
+ void processRanges(T& t) const
{
if (size() % 2) { //must be even number
throw InvalidArgumentException("SequenceNumberSet contains odd
number of elements");
}
for (SequenceNumberSet::const_iterator i = begin(); i != end(); i++) {
- SequenceNumber first = i->getValue();
- SequenceNumber last = (++i)->getValue();
+ SequenceNumber first = *(i);
+ SequenceNumber last = *(++i);
t(first, last);
}
}
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/AccumulatedAckTest.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/AccumulatedAckTest.cpp?rev=581957&r1=581956&r2=581957&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/AccumulatedAckTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/AccumulatedAckTest.cpp Thu Oct 4
09:06:05 2007
@@ -32,6 +32,7 @@
CPPUNIT_TEST_SUITE(AccumulatedAckTest);
CPPUNIT_TEST(testGeneral);
CPPUNIT_TEST(testCovers);
+ CPPUNIT_TEST(testUpdateFromCompletionData);
CPPUNIT_TEST(testCase1);
CPPUNIT_TEST(testCase2);
CPPUNIT_TEST(testCase3);
@@ -39,6 +40,7 @@
CPPUNIT_TEST(testConsolidation1);
CPPUNIT_TEST(testConsolidation2);
CPPUNIT_TEST(testConsolidation3);
+ CPPUNIT_TEST(testConsolidation4);
CPPUNIT_TEST_SUITE_END();
public:
@@ -97,6 +99,25 @@
CPPUNIT_ASSERT(!covers(ack, 10));
}
+ void testUpdateFromCompletionData()
+ {
+ AccumulatedAck ack(0);
+ SequenceNumber mark(2);
+ SequenceNumberSet ranges;
+ ranges.addRange(SequenceNumber(5), SequenceNumber(8));
+ ranges.addRange(SequenceNumber(10), SequenceNumber(15));
+ ranges.addRange(SequenceNumber(9), SequenceNumber(9));
+ ranges.addRange(SequenceNumber(3), SequenceNumber(4));
+
+ ack.update(mark, ranges);
+
+ for(int i = 0; i <= 15; i++) {
+ CPPUNIT_ASSERT(covers(ack, i));
+ }
+ CPPUNIT_ASSERT(!covers(ack, 16));
+ CPPUNIT_ASSERT_EQUAL((uint32_t) 15, ack.mark.getValue());
+ }
+
void testCase1()
{
AccumulatedAck ack(3);
@@ -203,6 +224,22 @@
update(ack, 1,15);
CPPUNIT_ASSERT_EQUAL((uint32_t) 15, ack.mark.getValue());
CPPUNIT_ASSERT_EQUAL((size_t) 0, ack.ranges.size());
+ }
+
+ void testConsolidation4()
+ {
+ AccumulatedAck ack(0);
+ ack.update(SequenceNumber(0), SequenceNumber(2));
+ ack.update(SequenceNumber(5), SequenceNumber(8));
+ ack.update(SequenceNumber(10), SequenceNumber(15));
+ ack.update(SequenceNumber(9), SequenceNumber(9));
+ ack.update(SequenceNumber(3), SequenceNumber(4));
+
+ for(int i = 0; i <= 15; i++) {
+ CPPUNIT_ASSERT(covers(ack, i));
+ }
+ CPPUNIT_ASSERT(!covers(ack, 16));
+ CPPUNIT_ASSERT_EQUAL((uint32_t) 15, ack.mark.getValue());
}
};
Modified: incubator/qpid/trunk/qpid/python/tests_0-10/message.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests_0-10/message.py?rev=581957&r1=581956&r2=581957&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests_0-10/message.py (original)
+++ incubator/qpid/trunk/qpid/python/tests_0-10/message.py Thu Oct 4 09:06:05
2007
@@ -612,6 +612,29 @@
for i in range (1, 11):
self.assertEquals("released message %s" % (i), queue.get(timeout =
1).content.body)
+ def test_ranged_ack(self):
+ """
+ Test acking of messages ranges
+ """
+ channel = self.channel
+ channel.queue_declare(queue = "q", exclusive=True)
+ for i in range (1, 11):
+ channel.message_transfer(content=Content(properties={'routing_key'
: "q"}, body = "message %s" % (i)))
+
+ channel.message_subscribe(queue = "q", destination = "a", confirm_mode
= 1)
+ channel.message_flow(unit = 0, value = 10, destination = "a")
+ channel.message_flow(unit = 1, value = 0xFFFFFFFF, destination = "a")
+ queue = self.client.queue("a")
+ for i in range (1, 11):
+ self.assertEquals("message %s" % (i), queue.get(timeout =
1).content.body)
+ self.assertEmpty(queue)
+
+ #ack all but the third message (command id 2)
+ channel.execution_complete(cumulative_execution_mark=0xFFFFFFFF,
ranged_execution_set=[0,1,3,6,7,7,8,9])
+ channel.message_recover()
+ self.assertEquals("message 3", queue.get(timeout = 1).content.body)
+ self.assertEmpty(queue)
+
def assertDataEquals(self, channel, msg, expected):
self.assertEquals(expected, msg.content.body)
