summaryrefslogtreecommitdiff
path: root/cpp/src/qpid/client/SubscriptionImpl.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/qpid/client/SubscriptionImpl.cpp')
-rw-r--r--cpp/src/qpid/client/SubscriptionImpl.cpp32
1 files changed, 28 insertions, 4 deletions
diff --git a/cpp/src/qpid/client/SubscriptionImpl.cpp b/cpp/src/qpid/client/SubscriptionImpl.cpp
index 684cca031a..0ccf5674fd 100644
--- a/cpp/src/qpid/client/SubscriptionImpl.cpp
+++ b/cpp/src/qpid/client/SubscriptionImpl.cpp
@@ -31,6 +31,9 @@ using framing::MessageAcquireResult;
SubscriptionImpl::SubscriptionImpl(SubscriptionManager& m, const std::string& q, const SubscriptionSettings& s, const std::string& n, MessageListener* l)
: manager(m), name(n), queue(q), settings(s), listener(l)
+{}
+
+void SubscriptionImpl::subscribe()
{
async(manager.getSession()).messageSubscribe(
arg::queue=queue,
@@ -79,11 +82,25 @@ void SubscriptionImpl::accept(const SequenceSet& messageIds) {
Mutex::ScopedLock l(lock);
manager.getSession().messageAccept(messageIds);
unaccepted.remove(messageIds);
- if (settings.autoComplete) {
+ switch (settings.completionMode) {
+ case COMPLETE_ON_ACCEPT:
+ manager.getSession().markCompleted(messageIds, true);
+ break;
+ case COMPLETE_ON_DELIVERY:
manager.getSession().sendCompletion();
+ break;
+ default://do nothing
+ break;
}
}
+void SubscriptionImpl::release(const SequenceSet& messageIds) {
+ Mutex::ScopedLock l(lock);
+ manager.getSession().messageRelease(messageIds);
+ if (settings.acceptMode == ACCEPT_MODE_EXPLICIT)
+ unaccepted.remove(messageIds);
+}
+
Session SubscriptionImpl::getSession() const { return manager.getSession(); }
SubscriptionManager&SubscriptionImpl:: getSubscriptionManager() const { return manager; }
@@ -102,16 +119,23 @@ void SubscriptionImpl::received(Message& m) {
listener->received(m);
}
- if (settings.autoComplete) {
+ if (settings.completionMode == COMPLETE_ON_DELIVERY) {
manager.getSession().markCompleted(m.getId(), false, false);
}
if (settings.autoAck) {
if (unaccepted.size() >= settings.autoAck) {
async(manager.getSession()).messageAccept(unaccepted);
- unaccepted.clear();
- if (settings.autoComplete) {
+ switch (settings.completionMode) {
+ case COMPLETE_ON_ACCEPT:
+ manager.getSession().markCompleted(unaccepted, true);
+ break;
+ case COMPLETE_ON_DELIVERY:
manager.getSession().sendCompletion();
+ break;
+ default://do nothing
+ break;
}
+ unaccepted.clear();
}
}
}