* added flag to SubscriptionSettings to control automatic completion of message

* removed automatic acquiring under autoAck mode
* added test for results from acquire requests
* added short txtest to the set of system tests run under make check



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

3 files changed, 38 insertions, 22 deletions

diff --git a/cpp/src/qpid/client/FlowControl.h b/cpp/src/qpid/client/FlowControl.h
index 0f5f8596ec..d2205aaa78 100644
--- a/cpp/src/qpid/client/FlowControl.h
+++ b/cpp/src/qpid/client/FlowControl.h
@@ -42,8 +42,9 @@ namespace client {
  * is renewed.
  *
  * In "window mode" credit is automatically renewed when a message is
- * accepted. In non-window mode credit is not automatically renewed,
- * it must be explicitly re-set (@see Subscription)
+ * completed (which by default happens when it is accepted). In
+ * non-window mode credit is not automatically renewed, it must be
+ * explicitly re-set (@see Subscription)
  */
 struct FlowControl {
     static const uint32_t UNLIMITED=0xFFFFFFFF;
diff --git a/cpp/src/qpid/client/SubscriptionImpl.cpp b/cpp/src/qpid/client/SubscriptionImpl.cpp
index 3363dda11f..684cca031a 100644
--- a/cpp/src/qpid/client/SubscriptionImpl.cpp
+++ b/cpp/src/qpid/client/SubscriptionImpl.cpp
@@ -27,6 +27,7 @@ namespace qpid {
 namespace client {
 
 using sys::Mutex;
+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)
@@ -68,16 +69,19 @@ SequenceSet SubscriptionImpl::getUnaccepted() const { Mutex::ScopedLock l(lock);
 
 void SubscriptionImpl::acquire(const SequenceSet& messageIds) {
     Mutex::ScopedLock l(lock);
-    manager.getSession().messageAcquire(messageIds);
-    unacquired.remove(messageIds);
+    MessageAcquireResult result = manager.getSession().messageAcquire(messageIds);
+    unacquired.remove(result.getTransfers());
     if (settings.acceptMode == ACCEPT_MODE_EXPLICIT)
-        unaccepted.add(messageIds);
+        unaccepted.add(result.getTransfers());
 }
 
 void SubscriptionImpl::accept(const SequenceSet& messageIds) {
     Mutex::ScopedLock l(lock);
     manager.getSession().messageAccept(messageIds);
     unaccepted.remove(messageIds);
+    if (settings.autoComplete) {
+        manager.getSession().sendCompletion();
+    }
 }
 
 Session SubscriptionImpl::getSession() const { return manager.getSession(); }
@@ -88,7 +92,6 @@ void SubscriptionImpl::cancel() { manager.cancel(name); }
 
 void SubscriptionImpl::received(Message& m) {
     Mutex::ScopedLock l(lock);
-    manager.getSession().markCompleted(m.getId(), false, false);        
     if (m.getMethod().getAcquireMode() == ACQUIRE_MODE_NOT_ACQUIRED) 
         unacquired.add(m.getId());
     else if (m.getMethod().getAcceptMode() == ACCEPT_MODE_EXPLICIT)
@@ -99,15 +102,16 @@ void SubscriptionImpl::received(Message& m) {
         listener->received(m);
     }
 
+    if (settings.autoComplete) {
+        manager.getSession().markCompleted(m.getId(), false, false);
+    }
     if (settings.autoAck) {
-        if (unacquired.size() + unaccepted.size() >= settings.autoAck) {
-            if (unacquired.size()) {
-                async(manager.getSession()).messageAcquire(unacquired);
-                unaccepted.add(unacquired);
-                unaccepted.clear();
-            }
+        if (unaccepted.size() >= settings.autoAck) {
             async(manager.getSession()).messageAccept(unaccepted);
             unaccepted.clear();
+            if (settings.autoComplete) {
+                manager.getSession().sendCompletion();
+            }
         }
     }
 }
diff --git a/cpp/src/qpid/client/SubscriptionSettings.h b/cpp/src/qpid/client/SubscriptionSettings.h
index 924814c809..19fbc3486b 100644
--- a/cpp/src/qpid/client/SubscriptionSettings.h
+++ b/cpp/src/qpid/client/SubscriptionSettings.h
@@ -39,22 +39,33 @@ struct SubscriptionSettings
         FlowControl flow=FlowControl::unlimited(),
         AcceptMode accept=ACCEPT_MODE_EXPLICIT,
         AcquireMode acquire=ACQUIRE_MODE_PRE_ACQUIRED,
-        unsigned int autoAck_=1
-    ) : flowControl(flow), acceptMode(accept), acquireMode(acquire), autoAck(autoAck_) {}
+        unsigned int autoAck_=1,
+        bool autoComplete_=true
+    ) : flowControl(flow), acceptMode(accept), acquireMode(acquire), autoAck(autoAck_), autoComplete(autoComplete_) {}
                          
     FlowControl flowControl;    ///@< Flow control settings. @see FlowControl
     AcceptMode acceptMode;      ///@< ACCEPT_MODE_EXPLICIT or ACCEPT_MODE_NONE
     AcquireMode acquireMode;    ///@< ACQUIRE_MODE_PRE_ACQUIRED or ACQUIRE_MODE_NOT_ACQUIRED
 
-    /** Automatically acknowledge (acquire and accept) batches of autoAck messages.
-     * 0 means no automatic acknowledgement. What it means to "acknowledge" a message depends on
-     * acceptMode and acquireMode:
-     *  - ACCEPT_MODE_NONE and ACQUIRE_MODE_PRE_ACQUIRED: do nothing
-     *  - ACCEPT_MODE_NONE and ACQUIRE_MODE_NOT_ACQUIRED: send an "acquire" command
-     *  - ACCEPT_MODE_EXPLICIT and ACQUIRE_MODE_PRE_ACQUIRED: send "accept" command
-     *  - ACCEPT_MODE_EXPLICIT and ACQUIRE_MODE_NOT_ACQUIRED: send "acquire" and "accept" commands
-     */
+    /** Automatically acknowledge (accept) batches of autoAck
+     *  messages. 0 means no automatic acknowledgement. This has no
+     *  effect for messsages received for a subscription with
+     *  ACCEPT_MODE_NODE.*/
     unsigned int autoAck;
+    /**
+     * If set to true, messages will be marked as completed (in
+     * windowing mode, completion of a message will cause the credit
+     * used up by that message to be reallocated) once they have been
+     * received. The server will be explicitly notified of all
+     * completed messages when the next accept is sent through the
+     * subscription (either explictly or through autAck). However the
+     * server may also periodically request information on the
+     * completed messages.
+     * 
+     * If set to false the application is responsible for completing
+     * messages (@see Session::markCompleted()).
+     */
+    bool autoComplete;
 };
 
 }} // namespace qpid::client