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/08/20 13:32:12 UTC
svn commit: r987459 - in /qpid/trunk/qpid/cpp: include/qpid/messaging/
src/qpid/client/amqp0_10/ src/qpid/messaging/ src/tests/
Author: gsim
Date: Fri Aug 20 11:32:11 2010
New Revision: 987459
URL: http://svn.apache.org/viewvc?rev=987459&view=rev
Log:
QPID-2807: Allow per message acknowledgement
Modified:
qpid/trunk/qpid/cpp/include/qpid/messaging/Session.h
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.h
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.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=987459&r1=987458&r2=987459&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/include/qpid/messaging/Session.h (original)
+++ qpid/trunk/qpid/cpp/include/qpid/messaging/Session.h Fri Aug 20 11:32:11 2010
@@ -76,6 +76,10 @@ class Session : public qpid::messaging::
*/
QPID_MESSAGING_EXTERN void acknowledge(bool sync=false);
/**
+ * Acknowledges the specified message.
+ */
+ QPID_MESSAGING_EXTERN void acknowledge(Message&, bool sync=false);
+ /**
* Rejects the specified message. This will prevent the message
* being redelivered. This must be called before the message is
* acknowledged.
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp?rev=987459&r1=987458&r2=987459&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp Fri Aug 20 11:32:11 2010
@@ -30,6 +30,14 @@ void AcceptTracker::State::accept()
unaccepted.clear();
}
+void AcceptTracker::State::accept(qpid::framing::SequenceNumber id)
+{
+ if (unaccepted.contains(id)) {
+ unaccepted.remove(id);
+ unconfirmed.add(id);
+ }
+}
+
void AcceptTracker::State::release()
{
unaccepted.clear();
@@ -63,6 +71,18 @@ void AcceptTracker::accept(qpid::client:
aggregateState.accept();
}
+void AcceptTracker::accept(qpid::framing::SequenceNumber id, qpid::client::AsyncSession& session)
+{
+ for (StateMap::iterator i = destinationState.begin(); i != destinationState.end(); ++i) {
+ i->second.accept(id);
+ }
+ Record record;
+ record.accepted.add(id);
+ record.status = session.messageAccept(record.accepted);
+ pending.push_back(record);
+ aggregateState.accept(id);
+}
+
void AcceptTracker::release(qpid::client::AsyncSession& session)
{
session.messageRelease(aggregateState.unaccepted);
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.h?rev=987459&r1=987458&r2=987459&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AcceptTracker.h Fri Aug 20 11:32:11 2010
@@ -42,6 +42,7 @@ class AcceptTracker
public:
void delivered(const std::string& destination, const qpid::framing::SequenceNumber& id);
void accept(qpid::client::AsyncSession&);
+ void accept(qpid::framing::SequenceNumber, qpid::client::AsyncSession&);
void release(qpid::client::AsyncSession&);
uint32_t acceptsPending();
uint32_t acceptsPending(const std::string& destination);
@@ -55,12 +56,13 @@ class AcceptTracker
*/
qpid::framing::SequenceSet unaccepted;
/**
- * ids of messages for which an accpet has been issued but not
+ * ids of messages for which an accept has been issued but not
* yet confirmed as completed
*/
qpid::framing::SequenceSet unconfirmed;
void accept();
+ void accept(qpid::framing::SequenceNumber);
void release();
uint32_t acceptsPending();
void completed(qpid::framing::SequenceSet&);
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp?rev=987459&r1=987458&r2=987459&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp Fri Aug 20 11:32:11 2010
@@ -144,6 +144,13 @@ void IncomingMessages::accept()
acceptTracker.accept(session);
}
+void IncomingMessages::accept(qpid::framing::SequenceNumber id)
+{
+ sys::Mutex::ScopedLock l(lock);
+ acceptTracker.accept(id, session);
+}
+
+
void IncomingMessages::releaseAll()
{
{
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.h?rev=987459&r1=987458&r2=987459&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.h Fri Aug 20 11:32:11 2010
@@ -72,6 +72,7 @@ class IncomingMessages
bool get(Handler& handler, qpid::sys::Duration timeout);
bool getNextDestination(std::string& destination, qpid::sys::Duration timeout);
void accept();
+ void accept(qpid::framing::SequenceNumber id);
void releaseAll();
void releasePending(const std::string& destination);
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=987459&r1=987458&r2=987459&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 Fri Aug 20 11:32:11 2010
@@ -112,6 +112,15 @@ void SessionImpl::release(qpid::messagin
execute1<Release>(m);
}
+void SessionImpl::acknowledge(qpid::messaging::Message& m)
+{
+ //Should probably throw an exception on failure here, or indicate
+ //it through a return type at least. Failure means that the
+ //message may be redelivered; i.e. the application cannot delete
+ //any state necessary for preventing reprocessing of the message
+ execute1<Acknowledge1>(m);
+}
+
void SessionImpl::close()
{
if (hasError()) {
@@ -456,6 +465,12 @@ void SessionImpl::acknowledgeImpl()
if (!transactional) incoming.accept();
}
+void SessionImpl::acknowledgeImpl(qpid::messaging::Message& m)
+{
+ ScopedLock l(lock);
+ if (!transactional) incoming.accept(MessageImplAccess::get(m).getInternalId());
+}
+
void SessionImpl::rejectImpl(qpid::messaging::Message& m)
{
SequenceSet set;
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=987459&r1=987458&r2=987459&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 Fri Aug 20 11:32:11 2010
@@ -63,6 +63,7 @@ class SessionImpl : public qpid::messagi
void acknowledge(bool sync);
void reject(qpid::messaging::Message&);
void release(qpid::messaging::Message&);
+ void acknowledge(qpid::messaging::Message& msg);
void close();
void sync(bool block);
qpid::messaging::Sender createSender(const qpid::messaging::Address& address);
@@ -137,6 +138,7 @@ class SessionImpl : public qpid::messagi
void commitImpl();
void rollbackImpl();
void acknowledgeImpl();
+ void acknowledgeImpl(qpid::messaging::Message&);
void rejectImpl(qpid::messaging::Message&);
void releaseImpl(qpid::messaging::Message&);
void closeImpl();
@@ -200,6 +202,14 @@ class SessionImpl : public qpid::messagi
Release(SessionImpl& i, qpid::messaging::Message& m) : Command(i), message(m) {}
void operator()() { impl.releaseImpl(message); }
};
+
+ struct Acknowledge1 : Command
+ {
+ qpid::messaging::Message& message;
+
+ Acknowledge1(SessionImpl& i, qpid::messaging::Message& m) : Command(i), message(m) {}
+ void operator()() { impl.acknowledgeImpl(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=987459&r1=987458&r2=987459&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/Session.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/Session.cpp Fri Aug 20 11:32:11 2010
@@ -39,6 +39,7 @@ Session& Session::operator=(const Sessio
void Session::commit() { impl->commit(); }
void Session::rollback() { impl->rollback(); }
void Session::acknowledge(bool sync) { impl->acknowledge(sync); }
+void Session::acknowledge(Message& m, bool s) { impl->acknowledge(m); if (s) sync(true); }
void Session::reject(Message& m) { impl->reject(m); }
void Session::release(Message& m) { impl->release(m); }
void Session::close() { impl->close(); }
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=987459&r1=987458&r2=987459&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/SessionImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/SessionImpl.h Fri Aug 20 11:32:11 2010
@@ -41,6 +41,7 @@ class SessionImpl : public virtual qpid:
virtual void commit() = 0;
virtual void rollback() = 0;
virtual void acknowledge(bool sync) = 0;
+ virtual void acknowledge(Message&) = 0;
virtual void reject(Message&) = 0;
virtual void release(Message&) = 0;
virtual void close() = 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=987459&r1=987458&r2=987459&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp Fri Aug 20 11:32:11 2010
@@ -791,6 +791,57 @@ QPID_AUTO_TEST_CASE(testExceptionOnClose
BOOST_CHECK_THROW(connection.createSession(), MessagingException);
}
+QPID_AUTO_TEST_CASE(testAcknowledge)
+{
+ QueueFixture fix;
+ Sender sender = fix.session.createSender(fix.queue);
+ const uint count(20);
+ for (uint i = 0; i < count; ++i) {
+ sender.send(Message((boost::format("Message_%1%") % (i+1)).str()));
+ }
+
+ Session other = fix.connection.createSession();
+ Receiver receiver = other.createReceiver(fix.queue);
+ std::vector<Message> messages;
+ for (uint i = 0; i < count; ++i) {
+ Message msg = receiver.fetch();
+ BOOST_CHECK_EQUAL(msg.getContent(), (boost::format("Message_%1%") % (i+1)).str());
+ messages.push_back(msg);
+ }
+ const uint batch(10); //acknowledge first 10 messages only
+ for (uint i = 0; i < batch; ++i) {
+ other.acknowledge(messages[i]);
+ }
+ messages.clear();
+ other.sync();
+ other.close();
+
+ other = fix.connection.createSession();
+ receiver = other.createReceiver(fix.queue);
+ for (uint i = 0; i < (count-batch); ++i) {
+ Message msg = receiver.fetch();
+ BOOST_CHECK_EQUAL(msg.getContent(), (boost::format("Message_%1%") % (i+1+batch)).str());
+ if (i % 2) other.acknowledge(msg); //acknowledge every other message
+ }
+ other.sync();
+ other.close();
+
+ //check unacknowledged messages are still enqueued
+ other = fix.connection.createSession();
+ receiver = other.createReceiver(fix.queue);
+ for (uint i = 0; i < ((count-batch)/2); ++i) {
+ Message msg = receiver.fetch();
+ BOOST_CHECK_EQUAL(msg.getContent(), (boost::format("Message_%1%") % ((i*2)+1+batch)).str());
+ }
+ other.acknowledge();//acknowledge all messages
+ other.sync();
+ other.close();
+
+ Message m;
+ //check queue is empty
+ BOOST_CHECK(!fix.session.createReceiver(fix.queue).fetch(m, Duration::IMMEDIATE));
+}
+
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