You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2010/02/15 19:36:22 UTC
svn commit: r910289 - in /qpid/trunk/qpid/cpp/src:
qpid/broker/DeliveryRecord.cpp qpid/broker/Message.cpp
qpid/broker/Message.h tests/ClientSessionTest.cpp
Author: kpvdr
Date: Mon Feb 15 18:36:21 2010
New Revision: 910289
URL: http://svn.apache.org/viewvc?rev=910289&view=rev
Log:
Added handling for adjusting TTL on outgoing message based on how long a message has been on the queue. Added a new TTL test as ClientSessionTest.testTtl.
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Message.h
qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp?rev=910289&r1=910288&r2=910289&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp Mon Feb 15 18:36:21 2010
@@ -76,6 +76,7 @@
if (msg.payload->getRedelivered()){
msg.payload->getProperties<framing::DeliveryProperties>()->setRedelivered(true);
}
+ msg.payload->adjustTtl();
framing::AMQFrame method((framing::MessageTransferBody(framing::ProtocolVersion(), tag, acceptExpected ? 0 : 1, acquired ? 0 : 1)));
method.setEof(false);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp?rev=910289&r1=910288&r2=910289&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp Mon Feb 15 18:36:21 2010
@@ -361,6 +361,16 @@
}
}
+void Message::adjustTtl()
+{
+ DeliveryProperties* props = getProperties<DeliveryProperties>();
+ if (props->getTtl()) {
+ sys::Mutex::ScopedLock l(lock);
+ sys::Duration d(sys::AbsTime::now(), getExpiration());
+ props->setTtl(int64_t(d) > 0 ? int64_t(d)/1000000 : 1); // convert from ns to ms; set to 1 if expired
+ }
+}
+
void Message::setExpiryPolicy(const boost::intrusive_ptr<ExpiryPolicy>& e) {
expiryPolicy = e;
if (expiryPolicy)
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Message.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Message.h?rev=910289&r1=910288&r2=910289&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Message.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Message.h Mon Feb 15 18:36:21 2010
@@ -81,6 +81,7 @@
void setExpiryPolicy(const boost::intrusive_ptr<ExpiryPolicy>& e);
bool hasExpired();
sys::AbsTime getExpiration() const { return expiration; }
+ void adjustTtl();
framing::FrameSet& getFrames() { return frames; }
const framing::FrameSet& getFrames() const { return frames; }
Modified: qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp?rev=910289&r1=910288&r2=910289&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp Mon Feb 15 18:36:21 2010
@@ -642,6 +642,36 @@
BOOST_CHECK_THROW(queue.get(1*qpid::sys::TIME_SEC), qpid::framing::ResourceDeletedException);
}
+QPID_AUTO_TEST_CASE(testTtl)
+{
+ const uint64_t ms = 1000ULL; // convert sec to ms
+ const uint64_t us = 1000ULL * 1000ULL; // convert sec to us
+
+ ClientSessionFixture fix;
+ fix.session.queueDeclare(arg::queue="ttl-test", arg::exclusive=true, arg::autoDelete=true);
+ Message msg1 = Message("AAA", "ttl-test");
+ uint64_t ttl = 2 * ms; // 2 sec
+ msg1.getDeliveryProperties().setTtl(ttl);
+ Connection c = fix.session.getConnection();
+ Session s = c.newSession();
+ s.messageTransfer(arg::content=msg1);
+
+ Message msg2 = Message("BBB", "ttl-test");
+ ttl = 10 * ms; // 10 sec
+ msg2.getDeliveryProperties().setTtl(ttl);
+ s.messageTransfer(arg::content=msg2);
+
+ ::usleep(5 * us); // 5 sec
+
+ // Message "AAA" should be expired and never be delivered
+ // Check "BBB" has ttl somewhere between 1 and 5 secs
+ Message got;
+ BOOST_CHECK(fix.subs.get(got, "ttl-test"));
+ BOOST_CHECK_EQUAL("BBB", got.getData());
+ BOOST_CHECK(got.getDeliveryProperties().getTtl() > 1 * ms);
+ BOOST_CHECK(got.getDeliveryProperties().getTtl() < ttl - (5 * ms));
+}
+
QPID_AUTO_TEST_SUITE_END()
}} // namespace qpid::tests
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org