From 17716fee99670e49a1c3526a44c40d15757d94e3 Mon Sep 17 00:00:00 2001
From: Alan Conway <aconway@apache.org>
Date: Thu, 6 Nov 2008 22:40:57 +0000
Subject: Add Message callbacks for async completion. Add unit test for async
 completion. Add sync parameter to generated session functions, defaults as
 before but allows greater control.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@711998 13f79535-47bb-0310-9956-ffa450edef68
---
 cpp/src/qpid/broker/Message.cpp | 27 +++++++++++++++++++++++----
 1 file changed, 23 insertions(+), 4 deletions(-)

(limited to 'cpp/src/qpid/broker/Message.cpp')

diff --git a/cpp/src/qpid/broker/Message.cpp b/cpp/src/qpid/broker/Message.cpp
index d2c5682359..89c647358a 100644
--- a/cpp/src/qpid/broker/Message.cpp
+++ b/cpp/src/qpid/broker/Message.cpp
@@ -45,9 +45,10 @@ namespace broker {
 
 TransferAdapter Message::TRANSFER;
 
-  Message::Message(const framing::SequenceNumber& id) : frames(id), persistenceId(0), redelivered(false), loaded(false),
-                                             staged(false), forcePersistentPolicy(false), publisher(0), adapter(0), 
-                                             expiration(FAR_FUTURE) {}
+Message::Message(const framing::SequenceNumber& id) :
+    frames(id), persistenceId(0), redelivered(false), loaded(false),
+    staged(false), forcePersistentPolicy(false), publisher(0), adapter(0), 
+    expiration(FAR_FUTURE), enqueueCallback(0), dequeueCallback(0) {}
 
 Message::~Message()
 {
@@ -268,7 +269,7 @@ bool Message::isContentLoaded() const
 
 namespace 
 {
-    const std::string X_QPID_TRACE("x-qpid.trace");
+const std::string X_QPID_TRACE("x-qpid.trace");
 }
 
 bool Message::isExcluded(const std::vector<std::string>& excludes) const
@@ -341,4 +342,22 @@ void Message::setReplacementMessage(boost::intrusive_ptr<Message> msg, const Que
     replacement[qfor] = msg;
 }
 
+void Message::allEnqueuesComplete() {
+    MessageCallback* cb = 0;
+    {
+        sys::Mutex::ScopedLock l(lock);
+        swap(cb, enqueueCallback);
+    }
+    if (cb && *cb) (*cb)(*this);
+}
+
+void Message::allDequeuesComplete() {
+    MessageCallback* cb = 0;
+    {
+        sys::Mutex::ScopedLock l(lock);
+        swap(cb, dequeueCallback);
+    }
+    if (cb && *cb) (*cb)(*this);
+}
+
 }} // namespace qpid::broker
-- 
cgit v1.2.1