diff options
| -rw-r--r-- | cpp/src/qpid/broker/DeliveryRecord.cpp | 1 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/Message.cpp | 10 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/Message.h | 1 | ||||
| -rw-r--r-- | cpp/src/tests/ClientSessionTest.cpp | 30 |
4 files changed, 42 insertions, 0 deletions
diff --git a/cpp/src/qpid/broker/DeliveryRecord.cpp b/cpp/src/qpid/broker/DeliveryRecord.cpp index 22ec5e86a0..64db84b6ec 100644 --- a/cpp/src/qpid/broker/DeliveryRecord.cpp +++ b/cpp/src/qpid/broker/DeliveryRecord.cpp @@ -76,6 +76,7 @@ void DeliveryRecord::deliver(framing::FrameHandler& h, DeliveryId deliveryId, ui 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); diff --git a/cpp/src/qpid/broker/Message.cpp b/cpp/src/qpid/broker/Message.cpp index 58de0cc9dc..ff4a37c88f 100644 --- a/cpp/src/qpid/broker/Message.cpp +++ b/cpp/src/qpid/broker/Message.cpp @@ -361,6 +361,16 @@ void Message::setTimestamp(const boost::intrusive_ptr<ExpiryPolicy>& e) } } +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) diff --git a/cpp/src/qpid/broker/Message.h b/cpp/src/qpid/broker/Message.h index becde0f872..0a7772040b 100644 --- a/cpp/src/qpid/broker/Message.h +++ b/cpp/src/qpid/broker/Message.h @@ -81,6 +81,7 @@ public: 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; } diff --git a/cpp/src/tests/ClientSessionTest.cpp b/cpp/src/tests/ClientSessionTest.cpp index e8cdb1f232..596f3282ce 100644 --- a/cpp/src/tests/ClientSessionTest.cpp +++ b/cpp/src/tests/ClientSessionTest.cpp @@ -642,6 +642,36 @@ QPID_AUTO_TEST_CASE(testQueueDeleted) 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 |