Author: gsim
Date: Mon Dec 11 02:44:03 2006
New Revision: 485594
URL: http://svn.apache.org/viewvc?view=rev&rev=485594
Log:
Allow xid to be associated with publication and acknowledgements.
Modified:
incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.cpp
incubator/qpid/trunk/qpid/cpp/lib/broker/DeliveryRecord.cpp
incubator/qpid/trunk/qpid/cpp/lib/broker/DeliveryRecord.h
incubator/qpid/trunk/qpid/cpp/lib/broker/TxAck.cpp
incubator/qpid/trunk/qpid/cpp/lib/broker/TxAck.h
incubator/qpid/trunk/qpid/cpp/lib/broker/TxPublish.cpp
incubator/qpid/trunk/qpid/cpp/lib/broker/TxPublish.h
incubator/qpid/trunk/qpid/cpp/tests/TxAckTest.cpp
incubator/qpid/trunk/qpid/cpp/tests/TxPublishTest.cpp
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.cpp?view=diff&rev=485594&r1=485593&r2=485594
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.cpp Mon Dec 11
02:44:03 2006
@@ -204,7 +204,7 @@
throw InvalidAckException();
}else if(multiple){
ack_iterator end = ++i;
- for_each(unacked.begin(), end,
bind2nd(mem_fun_ref(&DeliveryRecord::discard),
static_cast<qpid::broker::TransactionContext*>(0)));
+ for_each(unacked.begin(), end,
mem_fun_ref(&DeliveryRecord::discard));
unacked.erase(unacked.begin(), end);
//recalculate the prefetch:
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/DeliveryRecord.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/DeliveryRecord.cpp?view=diff&rev=485594&r1=485593&r2=485594
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/DeliveryRecord.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/DeliveryRecord.cpp Mon Dec 11
02:44:03 2006
@@ -42,8 +42,12 @@
pull(true){}
-void DeliveryRecord::discard(TransactionContext* ctxt) const{
- queue->dequeue(ctxt, msg, 0);
+void DeliveryRecord::discard(TransactionContext* ctxt, const std::string*
const xid) const{
+ queue->dequeue(ctxt, msg, xid);
+}
+
+void DeliveryRecord::discard() const{
+ discard(0, 0);
}
bool DeliveryRecord::matches(u_int64_t tag) const{
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/DeliveryRecord.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/DeliveryRecord.h?view=diff&rev=485594&r1=485593&r2=485594
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/DeliveryRecord.h (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/DeliveryRecord.h Mon Dec 11
02:44:03 2006
@@ -46,7 +46,8 @@
DeliveryRecord(Message::shared_ptr msg, Queue::shared_ptr queue,
const std::string consumerTag, const u_int64_t deliveryTag);
DeliveryRecord(Message::shared_ptr msg, Queue::shared_ptr queue,
const u_int64_t deliveryTag);
- void discard(TransactionContext* ctxt = 0) const;
+ void discard() const;
+ void discard(TransactionContext* ctxt, const std::string* const
xid) const;
bool matches(u_int64_t tag) const;
bool coveredBy(const AccumulatedAck* const range) const;
void requeue() const;
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/TxAck.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/TxAck.cpp?view=diff&rev=485594&r1=485593&r2=485594
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/TxAck.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/TxAck.cpp Mon Dec 11 02:44:03 2006
@@ -25,7 +25,8 @@
using std::mem_fun_ref;
using namespace qpid::broker;
-TxAck::TxAck(AccumulatedAck& _acked, std::list<DeliveryRecord>& _unacked) :
acked(_acked), unacked(_unacked){
+TxAck::TxAck(AccumulatedAck& _acked, std::list<DeliveryRecord>& _unacked,
const std::string* const _xid) :
+ acked(_acked), unacked(_unacked), xid(_xid){
}
@@ -34,10 +35,9 @@
//dequeue all acked messages from their queues
for (ack_iterator i = unacked.begin(); i != unacked.end(); i++) {
if (i->coveredBy(&acked)) {
- i->discard(ctxt);
+ i->discard(ctxt, xid);
}
}
- //for_each(unacked.begin(), unacked.end(),
bind2nd(mem_fun_ref(&DeliveryRecord::discardIfCoveredBy), &acked));
return true;
}catch(...){
std::cout << "TxAck::prepare() - Failed to prepare" << std::endl;
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/TxAck.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/TxAck.h?view=diff&rev=485594&r1=485593&r2=485594
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/TxAck.h (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/TxAck.h Mon Dec 11 02:44:03 2006
@@ -37,13 +37,15 @@
class TxAck : public TxOp{
AccumulatedAck& acked;
std::list<DeliveryRecord>& unacked;
+ const std::string* const xid;
+
public:
/**
* @param acked a representation of the accumulation of
* acks received
* @param unacked the record of delivered messages
*/
- TxAck(AccumulatedAck& acked, std::list<DeliveryRecord>& unacked);
+ TxAck(AccumulatedAck& acked, std::list<DeliveryRecord>& unacked,
const std::string* const xid = 0);
virtual bool prepare(TransactionContext* ctxt) throw();
virtual void commit() throw();
virtual void rollback() throw();
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/TxPublish.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/TxPublish.cpp?view=diff&rev=485594&r1=485593&r2=485594
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/TxPublish.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/TxPublish.cpp Mon Dec 11 02:44:03
2006
@@ -22,11 +22,11 @@
using namespace qpid::broker;
-TxPublish::TxPublish(Message::shared_ptr _msg) : msg(_msg) {}
+TxPublish::TxPublish(Message::shared_ptr _msg, const std::string* const _xid)
: msg(_msg), xid(_xid) {}
bool TxPublish::prepare(TransactionContext* ctxt) throw(){
try{
- for_each(queues.begin(), queues.end(), Prepare(ctxt, msg, 0));
+ for_each(queues.begin(), queues.end(), Prepare(ctxt, msg, xid));
return true;
}catch(...){
std::cout << "TxPublish::prepare() - Failed to prepare" << std::endl;
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/TxPublish.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/TxPublish.h?view=diff&rev=485594&r1=485593&r2=485594
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/TxPublish.h (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/TxPublish.h Mon Dec 11 02:44:03
2006
@@ -60,10 +60,11 @@
};
Message::shared_ptr msg;
+ const std::string* const xid;
std::list<Queue::shared_ptr> queues;
public:
- TxPublish(Message::shared_ptr msg);
+ TxPublish(Message::shared_ptr msg, const std::string* const xid =
0);
virtual bool prepare(TransactionContext* ctxt) throw();
virtual void commit() throw();
virtual void rollback() throw();
Modified: incubator/qpid/trunk/qpid/cpp/tests/TxAckTest.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/tests/TxAckTest.cpp?view=diff&rev=485594&r1=485593&r2=485594
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/tests/TxAckTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/tests/TxAckTest.cpp Mon Dec 11 02:44:03 2006
@@ -37,11 +37,11 @@
class TestMessageStore : public NullMessageStore
{
public:
- vector<Message*> dequeued;
+ vector< std::pair<Message*, const string*> > dequeued;
- void dequeue(TransactionContext*, Message* const msg, const Queue&
/*queue*/, const string * const /*xid*/)
+ void dequeue(TransactionContext*, Message* const msg, const Queue&
/*queue*/, const string * const xid)
{
- dequeued.push_back(msg);
+ dequeued.push_back(std::pair<Message*, const string*>(msg, xid));
}
TestMessageStore() : NullMessageStore(false) {}
@@ -49,6 +49,7 @@
};
CPPUNIT_TEST_SUITE(TxAckTest);
+ CPPUNIT_TEST(testPrepare2pc);
CPPUNIT_TEST(testPrepare);
CPPUNIT_TEST(testCommit);
CPPUNIT_TEST_SUITE_END();
@@ -60,11 +61,12 @@
vector<Message::shared_ptr> messages;
list<DeliveryRecord> deliveries;
TxAck op;
+ std::string xid;
public:
- TxAckTest() : queue(new Queue("my_queue", false, &store, 0)), op(acked,
deliveries)
+ TxAckTest() : queue(new Queue("my_queue", false, &store, 0)), op(acked,
deliveries, &xid)
{
for(int i = 0; i < 10; i++){
Message::shared_ptr msg(new Message(0, "exchange", "routing_key",
false, false));
@@ -86,13 +88,20 @@
op.prepare(0);
CPPUNIT_ASSERT_EQUAL((size_t) 7, store.dequeued.size());
CPPUNIT_ASSERT_EQUAL((size_t) 10, deliveries.size());
- CPPUNIT_ASSERT_EQUAL(messages[0].get(), store.dequeued[0]);//msg 1
- CPPUNIT_ASSERT_EQUAL(messages[1].get(), store.dequeued[1]);//msg 2
- CPPUNIT_ASSERT_EQUAL(messages[2].get(), store.dequeued[2]);//msg 3
- CPPUNIT_ASSERT_EQUAL(messages[3].get(), store.dequeued[3]);//msg 4
- CPPUNIT_ASSERT_EQUAL(messages[4].get(), store.dequeued[4]);//msg 5
- CPPUNIT_ASSERT_EQUAL(messages[6].get(), store.dequeued[5]);//msg 7
- CPPUNIT_ASSERT_EQUAL(messages[8].get(), store.dequeued[6]);//msg 9
+ int dequeued[] = {0, 1, 2, 3, 4, 6, 8};
+ for (int i = 0; i < 7; i++) {
+ CPPUNIT_ASSERT_EQUAL(messages[dequeued[i]].get(),
store.dequeued[i].first);
+ }
+ }
+
+ void testPrepare2pc()
+ {
+ xid = "abcdefg";
+ testPrepare();
+ const string expected(xid);
+ for (int i = 0; i < 7; i++) {
+ CPPUNIT_ASSERT_EQUAL(expected, *store.dequeued[i].second);
+ }
}
void testCommit()
Modified: incubator/qpid/trunk/qpid/cpp/tests/TxPublishTest.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/tests/TxPublishTest.cpp?view=diff&rev=485594&r1=485593&r2=485594
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/tests/TxPublishTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/tests/TxPublishTest.cpp Mon Dec 11 02:44:03
2006
@@ -34,15 +34,22 @@
class TxPublishTest : public CppUnit::TestCase
{
+ struct Triple
+ {
+ string first;
+ Message* second;
+ const string* third;
+ };
class TestMessageStore : public NullMessageStore
{
public:
- vector< pair<string, Message*> > enqueued;
+ vector<Triple> enqueued;
- void enqueue(TransactionContext*, Message* const msg, const Queue&
queue, const string * const /*xid*/)
+ void enqueue(TransactionContext*, Message* const msg, const Queue&
queue, const string * const xid)
{
- enqueued.push_back(pair<string, Message*>(queue.getName(),msg));
+ Triple args = {queue.getName(), msg, xid};
+ enqueued.push_back(args);
}
//dont care about any of the other methods:
@@ -52,6 +59,7 @@
CPPUNIT_TEST_SUITE(TxPublishTest);
CPPUNIT_TEST(testPrepare);
+ CPPUNIT_TEST(testPrepare2pc);
CPPUNIT_TEST(testCommit);
CPPUNIT_TEST_SUITE_END();
@@ -61,14 +69,14 @@
Queue::shared_ptr queue2;
Message::shared_ptr const msg;
TxPublish op;
-
+ string xid;
public:
TxPublishTest() : queue1(new Queue("queue1", false, &store, 0)),
queue2(new Queue("queue2", false, &store, 0)),
msg(new Message(0, "exchange", "routing_key", false,
false)),
- op(msg)
+ op(msg, &xid)
{
msg->setHeader(AMQHeaderBody::shared_ptr(new AMQHeaderBody(BASIC)));
msg->getHeaderProperties()->setDeliveryMode(PERSISTENT);
@@ -85,6 +93,15 @@
CPPUNIT_ASSERT_EQUAL(msg.get(), store.enqueued[0].second);
CPPUNIT_ASSERT_EQUAL(string("queue2"), store.enqueued[1].first);
CPPUNIT_ASSERT_EQUAL(msg.get(), store.enqueued[1].second);
+ }
+
+ void testPrepare2pc()
+ {
+ xid = "abcde";
+ const string expected(xid);
+ testPrepare();
+ CPPUNIT_ASSERT_EQUAL(expected, *store.enqueued[0].third);
+ CPPUNIT_ASSERT_EQUAL(expected, *store.enqueued[1].third);
}
void testCommit()
