QPID-664: added param to control synchronous send and acknowledge

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

4 files changed, 6 insertions, 4 deletions

diff --git a/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp b/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp
index 9bb785e13f..e8c106976f 100644
--- a/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp
+++ b/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp
@@ -34,7 +34,7 @@ SenderImpl::SenderImpl(SessionImpl& _parent, const std::string& _name,
     parent(&_parent), name(_name), address(_address), state(UNRESOLVED),
     capacity(50), window(0), flushed(false), unreliable(AddressResolution::is_unreliable(address)) {}
 
-void SenderImpl::send(const qpid::messaging::Message& message) 
+void SenderImpl::send(const qpid::messaging::Message& message, bool sync) 
 {
     if (unreliable) {
         UnreliableSend f(*this, &message);
@@ -43,6 +43,7 @@ void SenderImpl::send(const qpid::messaging::Message& message)
         Send f(*this, &message);
         while (f.repeat) parent->execute(f);
     }
+    if (sync) parent->sync();
 }
 
 void SenderImpl::close()
diff --git a/cpp/src/qpid/client/amqp0_10/SenderImpl.h b/cpp/src/qpid/client/amqp0_10/SenderImpl.h
index a99d4bd294..7ea68fd187 100644
--- a/cpp/src/qpid/client/amqp0_10/SenderImpl.h
+++ b/cpp/src/qpid/client/amqp0_10/SenderImpl.h
@@ -48,7 +48,7 @@ class SenderImpl : public qpid::messaging::SenderImpl
 
     SenderImpl(SessionImpl& parent, const std::string& name, 
                const qpid::messaging::Address& address);
-    void send(const qpid::messaging::Message&);
+    void send(const qpid::messaging::Message&, bool sync);
     void close();
     void setCapacity(uint32_t);
     uint32_t getCapacity();
diff --git a/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp b/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
index 209ab93909..bb0a735fde 100644
--- a/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
+++ b/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
@@ -75,13 +75,14 @@ void SessionImpl::rollback()
     execute<Rollback>();
 }
 
-void SessionImpl::acknowledge()
+void SessionImpl::acknowledge(bool sync_)
 {
     //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
     execute<Acknowledge>();
+    if (sync_) sync();
 }
 
 void SessionImpl::reject(qpid::messaging::Message& m)
diff --git a/cpp/src/qpid/client/amqp0_10/SessionImpl.h b/cpp/src/qpid/client/amqp0_10/SessionImpl.h
index a616db6239..ab29d2c1e1 100644
--- a/cpp/src/qpid/client/amqp0_10/SessionImpl.h
+++ b/cpp/src/qpid/client/amqp0_10/SessionImpl.h
@@ -58,7 +58,7 @@ class SessionImpl : public qpid::messaging::SessionImpl
     SessionImpl(ConnectionImpl&, bool transactional);
     void commit();
     void rollback();
-    void acknowledge();
+    void acknowledge(bool sync);
     void reject(qpid::messaging::Message&);
     void close();
     void sync();