diff options
author | Gordon Sim <gsim@apache.org> | 2010-04-06 14:52:47 +0000 |
---|---|---|
committer | Gordon Sim <gsim@apache.org> | 2010-04-06 14:52:47 +0000 |
commit | 8fddccd7efb73cd25376f5492ddb09ad7fa09e3b (patch) | |
tree | 2c2e7045eb1a4733fbb24a4f7106e96dca945ae0 | |
parent | 48d21770c0c93551180043ddf27b1b7178382a4a (diff) | |
download | qpid-python-8fddccd7efb73cd25376f5492ddb09ad7fa09e3b.tar.gz |
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
-rw-r--r-- | cpp/examples/messaging/map_sender.cpp | 3 | ||||
-rw-r--r-- | cpp/examples/messaging/queue_sender.cpp | 3 | ||||
-rw-r--r-- | cpp/examples/messaging/topic_sender.cpp | 3 | ||||
-rw-r--r-- | cpp/include/qpid/messaging/Sender.h | 9 | ||||
-rw-r--r-- | cpp/include/qpid/messaging/Session.h | 5 | ||||
-rw-r--r-- | cpp/src/qpid/client/amqp0_10/SenderImpl.cpp | 3 | ||||
-rw-r--r-- | cpp/src/qpid/client/amqp0_10/SenderImpl.h | 2 | ||||
-rw-r--r-- | cpp/src/qpid/client/amqp0_10/SessionImpl.cpp | 3 | ||||
-rw-r--r-- | cpp/src/qpid/client/amqp0_10/SessionImpl.h | 2 | ||||
-rw-r--r-- | cpp/src/qpid/messaging/Sender.cpp | 2 | ||||
-rw-r--r-- | cpp/src/qpid/messaging/SenderImpl.h | 2 | ||||
-rw-r--r-- | cpp/src/qpid/messaging/Session.cpp | 2 | ||||
-rw-r--r-- | cpp/src/qpid/messaging/SessionImpl.h | 2 | ||||
-rw-r--r-- | cpp/src/tests/MessagingSessionTests.cpp | 12 |
14 files changed, 36 insertions, 17 deletions
diff --git a/cpp/examples/messaging/map_sender.cpp b/cpp/examples/messaging/map_sender.cpp index 8c85510717..8dba47ce63 100644 --- a/cpp/examples/messaging/map_sender.cpp +++ b/cpp/examples/messaging/map_sender.cpp @@ -56,8 +56,7 @@ int main(int argc, char** argv) { content["uuid"] = Uuid(true); encode(content, message); - sender.send(message); - session.sync(); + sender.send(message, true); connection.close(); return 0; diff --git a/cpp/examples/messaging/queue_sender.cpp b/cpp/examples/messaging/queue_sender.cpp index 0d86f024b3..f7f0e9bf6d 100644 --- a/cpp/examples/messaging/queue_sender.cpp +++ b/cpp/examples/messaging/queue_sender.cpp @@ -48,8 +48,7 @@ int main(int argc, char** argv) { } // And send a final message to indicate termination. - sender.send(Message("That's all, folks!")); - session.sync(); + sender.send(Message("That's all, folks!"), true); return 0; } catch(const std::exception& error) { std::cout << error.what() << std::endl; diff --git a/cpp/examples/messaging/topic_sender.cpp b/cpp/examples/messaging/topic_sender.cpp index 6b1f85ec6f..5dd593ff8a 100644 --- a/cpp/examples/messaging/topic_sender.cpp +++ b/cpp/examples/messaging/topic_sender.cpp @@ -66,8 +66,7 @@ int main(int argc, char** argv) { // And send a final message to indicate termination. Message message("That's all, folks!"); message.setSubject("control"); - sender.send(message); - session.sync(); + sender.send(message, true); connection.close(); return 0; } catch(const std::exception& error) { diff --git a/cpp/include/qpid/messaging/Sender.h b/cpp/include/qpid/messaging/Sender.h index 66a7b1fe37..7c4b68731e 100644 --- a/cpp/include/qpid/messaging/Sender.h +++ b/cpp/include/qpid/messaging/Sender.h @@ -45,9 +45,14 @@ class Sender : public qpid::messaging::Handle<SenderImpl> QPID_CLIENT_EXTERN Sender& operator=(const Sender&); /** - * Sends a message; will block if the pending == capacity + * Sends a message + * + * @param message the message to send + * @param sync if true the call will block until the server + * confirms receipt of the messages; if false will only block for + * available capacity (i.e. pending == capacity) */ - QPID_CLIENT_EXTERN void send(const Message& message); + QPID_CLIENT_EXTERN void send(const Message& message, bool sync=false); QPID_CLIENT_EXTERN void close(); /** diff --git a/cpp/include/qpid/messaging/Session.h b/cpp/include/qpid/messaging/Session.h index 66a4937e6b..9b373e2d8b 100644 --- a/cpp/include/qpid/messaging/Session.h +++ b/cpp/include/qpid/messaging/Session.h @@ -71,8 +71,11 @@ class Session : public qpid::messaging::Handle<SessionImpl> /** * Acknowledges all outstanding messages that have been received * by the application on this session. + * + * @param sync if true, blocks until the acknowledgement has been + * processed by the server */ - QPID_CLIENT_EXTERN void acknowledge(); + QPID_CLIENT_EXTERN void acknowledge(bool sync=false); /** * Rejects the specified message. This will prevent the message * being redelivered. 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(); diff --git a/cpp/src/qpid/messaging/Sender.cpp b/cpp/src/qpid/messaging/Sender.cpp index 2d5cfbcec5..b4c247d1d9 100644 --- a/cpp/src/qpid/messaging/Sender.cpp +++ b/cpp/src/qpid/messaging/Sender.cpp @@ -32,7 +32,7 @@ Sender::Sender(SenderImpl* impl) { PI::ctor(*this, impl); } Sender::Sender(const Sender& s) : qpid::messaging::Handle<SenderImpl>() { PI::copy(*this, s); } Sender::~Sender() { PI::dtor(*this); } Sender& Sender::operator=(const Sender& s) { return PI::assign(*this, s); } -void Sender::send(const Message& message) { impl->send(message); } +void Sender::send(const Message& message, bool sync) { impl->send(message, sync); } void Sender::close() { impl->close(); } void Sender::setCapacity(uint32_t c) { impl->setCapacity(c); } uint32_t Sender::getCapacity() { return impl->getCapacity(); } diff --git a/cpp/src/qpid/messaging/SenderImpl.h b/cpp/src/qpid/messaging/SenderImpl.h index 7653049c26..66651a16dc 100644 --- a/cpp/src/qpid/messaging/SenderImpl.h +++ b/cpp/src/qpid/messaging/SenderImpl.h @@ -33,7 +33,7 @@ class SenderImpl : public virtual qpid::RefCounted { public: virtual ~SenderImpl() {} - virtual void send(const Message& message) = 0; + virtual void send(const Message& message, bool sync) = 0; virtual void close() = 0; virtual void setCapacity(uint32_t) = 0; virtual uint32_t getCapacity() = 0; diff --git a/cpp/src/qpid/messaging/Session.cpp b/cpp/src/qpid/messaging/Session.cpp index 4d1e633a86..2c09bdf7a6 100644 --- a/cpp/src/qpid/messaging/Session.cpp +++ b/cpp/src/qpid/messaging/Session.cpp @@ -38,7 +38,7 @@ Session::~Session() { PI::dtor(*this); } Session& Session::operator=(const Session& s) { return PI::assign(*this, s); } void Session::commit() { impl->commit(); } void Session::rollback() { impl->rollback(); } -void Session::acknowledge() { impl->acknowledge(); } +void Session::acknowledge(bool sync) { impl->acknowledge(sync); } void Session::reject(Message& m) { impl->reject(m); } void Session::close() { impl->close(); } diff --git a/cpp/src/qpid/messaging/SessionImpl.h b/cpp/src/qpid/messaging/SessionImpl.h index 79f0d007b5..711acc324c 100644 --- a/cpp/src/qpid/messaging/SessionImpl.h +++ b/cpp/src/qpid/messaging/SessionImpl.h @@ -40,7 +40,7 @@ class SessionImpl : public virtual qpid::RefCounted virtual ~SessionImpl() {} virtual void commit() = 0; virtual void rollback() = 0; - virtual void acknowledge() = 0; + virtual void acknowledge(bool sync) = 0; virtual void reject(Message&) = 0; virtual void close() = 0; virtual void sync() = 0; diff --git a/cpp/src/tests/MessagingSessionTests.cpp b/cpp/src/tests/MessagingSessionTests.cpp index bdfb8b389c..e1c0a6bb16 100644 --- a/cpp/src/tests/MessagingSessionTests.cpp +++ b/cpp/src/tests/MessagingSessionTests.cpp @@ -231,6 +231,18 @@ QPID_AUTO_TEST_CASE(testSimpleSendReceive) BOOST_CHECK_EQUAL(in.getContent(), out.getContent()); } +QPID_AUTO_TEST_CASE(testSyncSendReceive) +{ + QueueFixture fix; + Sender sender = fix.session.createSender(fix.queue); + Message out("test-message"); + sender.send(out, true); + Receiver receiver = fix.session.createReceiver(fix.queue); + Message in = receiver.fetch(Duration::IMMEDIATE); + fix.session.acknowledge(true); + BOOST_CHECK_EQUAL(in.getContent(), out.getContent()); +} + QPID_AUTO_TEST_CASE(testSendReceiveHeaders) { QueueFixture fix; |