From 8fddccd7efb73cd25376f5492ddb09ad7fa09e3b Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Tue, 6 Apr 2010 14:52:47 +0000 Subject: 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 --- cpp/src/qpid/client/amqp0_10/SenderImpl.cpp | 3 ++- cpp/src/qpid/client/amqp0_10/SenderImpl.h | 2 +- cpp/src/qpid/client/amqp0_10/SessionImpl.cpp | 3 ++- cpp/src/qpid/client/amqp0_10/SessionImpl.h | 2 +- 4 files changed, 6 insertions(+), 4 deletions(-) (limited to 'cpp/src/qpid/client') 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(); } -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(); + 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(); -- cgit v1.2.1