QPID-664: added method to release messages

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@931275 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 47 insertions, 1 deletions

diff --git a/cpp/include/qpid/messaging/Session.h b/cpp/include/qpid/messaging/Session.h
index 9b373e2d8b..c20de9079b 100644
--- a/cpp/include/qpid/messaging/Session.h
+++ b/cpp/include/qpid/messaging/Session.h
@@ -78,9 +78,16 @@ class Session : public qpid::messaging::Handle<SessionImpl>
     QPID_CLIENT_EXTERN void acknowledge(bool sync=false);
     /**
      * Rejects the specified message. This will prevent the message
-     * being redelivered.
+     * being redelivered. This must be called before the message is
+     * acknowledged.
      */
     QPID_CLIENT_EXTERN void reject(Message&);
+    /**
+     * Releases the specified message. This will allow the broker to
+     * redeliver the message. This must be called before the message
+     * is acknowledged.
+     */
+    QPID_CLIENT_EXTERN void release(Message&);
 
     QPID_CLIENT_EXTERN void sync();
     QPID_CLIENT_EXTERN void flush();
diff --git a/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp b/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
index bb0a735fde..9efafb1d16 100644
--- a/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
+++ b/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
@@ -92,6 +92,11 @@ void SessionImpl::reject(qpid::messaging::Message& m)
     execute1<Reject>(m);
 }
 
+void SessionImpl::release(qpid::messaging::Message& m)
+{
+    execute1<Release>(m);
+}
+
 void SessionImpl::close()
 {
     //close all the senders and receivers (get copy of names and then
@@ -418,6 +423,13 @@ void SessionImpl::rejectImpl(qpid::messaging::Message& m)
     session.messageReject(set);
 }
 
+void SessionImpl::releaseImpl(qpid::messaging::Message& m)
+{
+    SequenceSet set;
+    set.add(MessageImplAccess::get(m).getInternalId());
+    session.messageRelease(set);
+}
+
 void SessionImpl::receiverCancelled(const std::string& name)
 {
     receivers.erase(name);
diff --git a/cpp/src/qpid/client/amqp0_10/SessionImpl.h b/cpp/src/qpid/client/amqp0_10/SessionImpl.h
index ab29d2c1e1..5f9b23fb14 100644
--- a/cpp/src/qpid/client/amqp0_10/SessionImpl.h
+++ b/cpp/src/qpid/client/amqp0_10/SessionImpl.h
@@ -60,6 +60,7 @@ class SessionImpl : public qpid::messaging::SessionImpl
     void rollback();
     void acknowledge(bool sync);
     void reject(qpid::messaging::Message&);
+    void release(qpid::messaging::Message&);
     void close();
     void sync();
     void flush();
@@ -123,6 +124,7 @@ class SessionImpl : public qpid::messaging::SessionImpl
     void rollbackImpl();
     void acknowledgeImpl();
     void rejectImpl(qpid::messaging::Message&);
+    void releaseImpl(qpid::messaging::Message&);
     void closeImpl();
     void syncImpl();
     void flushImpl();
@@ -177,6 +179,14 @@ class SessionImpl : public qpid::messaging::SessionImpl
         Reject(SessionImpl& i, qpid::messaging::Message& m) : Command(i), message(m) {}
         void operator()() { impl.rejectImpl(message); }
     };
+
+    struct Release : Command
+    {
+        qpid::messaging::Message& message;
+
+        Release(SessionImpl& i, qpid::messaging::Message& m) : Command(i), message(m) {}
+        void operator()() { impl.releaseImpl(message); }
+    };
     
     struct CreateSender;
     struct CreateReceiver;
diff --git a/cpp/src/qpid/messaging/Session.cpp b/cpp/src/qpid/messaging/Session.cpp
index 2c09bdf7a6..cc0b528777 100644
--- a/cpp/src/qpid/messaging/Session.cpp
+++ b/cpp/src/qpid/messaging/Session.cpp
@@ -40,6 +40,7 @@ void Session::commit() { impl->commit(); }
 void Session::rollback() { impl->rollback(); }
 void Session::acknowledge(bool sync) { impl->acknowledge(sync); }
 void Session::reject(Message& m) { impl->reject(m); }
+void Session::release(Message& m) { impl->release(m); }
 void Session::close() { impl->close(); }
 
 Sender Session::createSender(const Address& address)
diff --git a/cpp/src/qpid/messaging/SessionImpl.h b/cpp/src/qpid/messaging/SessionImpl.h
index 711acc324c..7f76f8556f 100644
--- a/cpp/src/qpid/messaging/SessionImpl.h
+++ b/cpp/src/qpid/messaging/SessionImpl.h
@@ -42,6 +42,7 @@ class SessionImpl : public virtual qpid::RefCounted
     virtual void rollback() = 0;
     virtual void acknowledge(bool sync) = 0;
     virtual void reject(Message&) = 0;
+    virtual void release(Message&) = 0;
     virtual void close() = 0;
     virtual void sync() = 0;
     virtual void flush() = 0;
diff --git a/cpp/src/tests/MessagingSessionTests.cpp b/cpp/src/tests/MessagingSessionTests.cpp
index e1c0a6bb16..6341ef3f28 100644
--- a/cpp/src/tests/MessagingSessionTests.cpp
+++ b/cpp/src/tests/MessagingSessionTests.cpp
@@ -857,6 +857,21 @@ QPID_AUTO_TEST_CASE(testTx)
     BOOST_CHECK(!receiver2.fetch(in, Duration::IMMEDIATE));
 }
 
+QPID_AUTO_TEST_CASE(testRelease)
+{
+    QueueFixture fix;
+    Sender sender = fix.session.createSender(fix.queue);
+    Message out("test-message");
+    sender.send(out, true);
+    Receiver receiver = fix.session.createReceiver(fix.queue);
+    Message m1 = receiver.fetch(Duration::IMMEDIATE);
+    fix.session.release(m1);
+    Message m2 = receiver.fetch(Duration::SECOND * 1);
+    BOOST_CHECK_EQUAL(m1.getContent(), out.getContent());
+    BOOST_CHECK_EQUAL(m1.getContent(), m2.getContent());
+    fix.session.acknowledge(true);
+}
+
 QPID_AUTO_TEST_SUITE_END()
 
 }} // namespace qpid::tests