You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2010/04/06 21:08:28 UTC
svn commit: r931275 - in /qpid/trunk/qpid/cpp: include/qpid/messaging/
src/qpid/client/amqp0_10/ src/qpid/messaging/ src/tests/
Author: gsim
Date: Tue Apr 6 19:08:28 2010
New Revision: 931275
URL: http://svn.apache.org/viewvc?rev=931275&view=rev
Log:
QPID-664: added method to release messages
Modified:
qpid/trunk/qpid/cpp/include/qpid/messaging/Session.h
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h
qpid/trunk/qpid/cpp/src/qpid/messaging/Session.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/SessionImpl.h
qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp
Modified: qpid/trunk/qpid/cpp/include/qpid/messaging/Session.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/include/qpid/messaging/Session.h?rev=931275&r1=931274&r2=931275&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/include/qpid/messaging/Session.h (original)
+++ qpid/trunk/qpid/cpp/include/qpid/messaging/Session.h Tue Apr 6 19:08:28 2010
@@ -78,9 +78,16 @@ class Session : public qpid::messaging::
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();
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp?rev=931275&r1=931274&r2=931275&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp Tue Apr 6 19:08:28 2010
@@ -92,6 +92,11 @@ void SessionImpl::reject(qpid::messaging
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::messa
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);
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h?rev=931275&r1=931274&r2=931275&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h Tue Apr 6 19:08:28 2010
@@ -60,6 +60,7 @@ class SessionImpl : public qpid::messagi
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::messagi
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::messagi
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;
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/Session.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/Session.cpp?rev=931275&r1=931274&r2=931275&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/Session.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/Session.cpp Tue Apr 6 19:08:28 2010
@@ -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)
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/SessionImpl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/SessionImpl.h?rev=931275&r1=931274&r2=931275&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/SessionImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/SessionImpl.h Tue Apr 6 19:08:28 2010
@@ -42,6 +42,7 @@ class SessionImpl : public virtual qpid:
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;
Modified: qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp?rev=931275&r1=931274&r2=931275&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp Tue Apr 6 19:08:28 2010
@@ -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
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org