QPID-920: send message-accept for acks (as well as completion)

* AckPolicy now maintains a set of transfered messages for cumulative accepts



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

5 files changed, 11 insertions, 8 deletions

diff --git a/cpp/src/tests/ClientSessionTest.cpp b/cpp/src/tests/ClientSessionTest.cpp
index 9b6e0dce21..a5f7b9d803 100644
--- a/cpp/src/tests/ClientSessionTest.cpp
+++ b/cpp/src/tests/ClientSessionTest.cpp
@@ -20,6 +20,7 @@
  */
 #include "unit_test.h"
 #include "BrokerFixture.h"
+#include "qpid/client/AckPolicy.h"
 #include "qpid/client/Dispatcher.h"
 #include "qpid/sys/Monitor.h"
 #include "qpid/sys/Thread.h"
@@ -124,7 +125,8 @@ QPID_AUTO_TEST_CASE(testTransfer)
     BOOST_CHECK(msg->isA<MessageTransferBody>());
     BOOST_CHECK_EQUAL(string("my-message"), msg->getContent());
     //confirm receipt:
-    fix.session.getExecution().markCompleted(msg->getId(), true, true);
+    AckPolicy autoAck;
+    autoAck.ack(Message(*msg), fix.session);
 }
 
 QPID_AUTO_TEST_CASE(testDispatcher)
diff --git a/cpp/src/tests/client_test.cpp b/cpp/src/tests/client_test.cpp
index 011dcd4678..d0da2ec8ca 100644
--- a/cpp/src/tests/client_test.cpp
+++ b/cpp/src/tests/client_test.cpp
@@ -124,10 +124,11 @@ int main(int argc, char** argv)
 	if (opts.trace) std::cout << "Subscribed to queue." << std::endl;
         FrameSet::shared_ptr incoming = session.get();
         if (incoming->isA<MessageTransferBody>()) {
-            Message msgIn(*incoming, session);
+            Message msgIn(*incoming);
             if (msgIn.getData() == msgOut.getData()) {
                 if (opts.trace) std::cout << "Received the exepected message." << std::endl;
-                msgIn.acknowledge();
+                session.messageAccept(SequenceSet(msgIn.getId()));
+                session.markCompleted(msgIn.getId(), true, true);
             } else {
                 print("Received an unexepected message: ", msgIn);
             }
diff --git a/cpp/src/tests/perftest.cpp b/cpp/src/tests/perftest.cpp
index 966d708ff6..231b25daa4 100644
--- a/cpp/src/tests/perftest.cpp
+++ b/cpp/src/tests/perftest.cpp
@@ -533,9 +533,9 @@ struct SubscribeThread : public Client {
                 size_t expect=0;
                 for (size_t i = 0; i < opts.subQuota; ++i) {
                     msg=lq.pop();
-		            if (opts.intervalSub) ::usleep(opts.intervalSub*1000);
+                    if (opts.intervalSub) ::usleep(opts.intervalSub*1000);
                     // TODO aconway 2007-11-23: check message order for. 
-                    // multiple publishers. Need an acorray of counters,
+                    // multiple publishers. Need an array of counters,
                     // one per publisher and a publisher ID in the
                     // message. Careful not to introduce a lot of overhead
                     // here, e.g. no std::map, std::string etc.
@@ -550,7 +550,7 @@ struct SubscribeThread : public Client {
                     }
                 }
                 if (opts.ack !=0)
-                    msg.acknowledge(); // Cumulative ack for final batch.
+                    subs.getAckPolicy().ackOutstanding(session); // Cumulative ack for final batch.
                 AbsTime end=now();
 
                 // Report to publisher.
diff --git a/cpp/src/tests/topic_listener.cpp b/cpp/src/tests/topic_listener.cpp
index 5208b67445..3dd042605e 100644
--- a/cpp/src/tests/topic_listener.cpp
+++ b/cpp/src/tests/topic_listener.cpp
@@ -159,7 +159,7 @@ void Listener::received(Message& message){
     if(!!type && StringValue("TERMINATION_REQUEST") == *type){
         shutdown();
     }else if(!!type && StringValue("REPORT_REQUEST") == *type){
-        message.acknowledge();//acknowledge everything upto this point
+        mgr.getAckPolicy().ackOutstanding(session);//acknowledge everything upto this point
         cout <<"Batch ended, sending report." << endl;
         //send a report:
         report();
diff --git a/cpp/src/tests/txtest.cpp b/cpp/src/tests/txtest.cpp
index 5030b24070..a8369df759 100644
--- a/cpp/src/tests/txtest.cpp
+++ b/cpp/src/tests/txtest.cpp
@@ -144,7 +144,7 @@ struct Transfer : public Client, public Runnable
                     out.getDeliveryProperties().setDeliveryMode(in.getDeliveryProperties().getDeliveryMode());
                     session.messageTransfer(arg::content=out, arg::acceptMode=1);
                 }
-                in.acknowledge();
+                lq.getAckPolicy().ackOutstanding(session);
                 session.txCommit();
             }
         } catch(const std::exception& e) {