summaryrefslogtreecommitdiff
path: root/cpp/src/qpid/broker/Message.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/qpid/broker/Message.cpp')
-rw-r--r--cpp/src/qpid/broker/Message.cpp36
1 files changed, 22 insertions, 14 deletions
diff --git a/cpp/src/qpid/broker/Message.cpp b/cpp/src/qpid/broker/Message.cpp
index 2d42c650ce..e2799b0bff 100644
--- a/cpp/src/qpid/broker/Message.cpp
+++ b/cpp/src/qpid/broker/Message.cpp
@@ -389,28 +389,36 @@ void Message::setReplacementMessage(boost::intrusive_ptr<Message> msg, const Que
}
void Message::allEnqueuesComplete() {
- MessageCallback* cb = 0;
- {
- sys::Mutex::ScopedLock l(lock);
- std::swap(cb, enqueueCallback);
- }
+ sys::Mutex::ScopedLock l(callbackLock);
+ MessageCallback* cb = enqueueCallback;
if (cb && *cb) (*cb)(intrusive_ptr<Message>(this));
}
void Message::allDequeuesComplete() {
- MessageCallback* cb = 0;
- {
- sys::Mutex::ScopedLock l(lock);
- std::swap(cb, dequeueCallback);
- }
+ sys::Mutex::ScopedLock l(callbackLock);
+ MessageCallback* cb = dequeueCallback;
if (cb && *cb) (*cb)(intrusive_ptr<Message>(this));
}
-void Message::setEnqueueCompleteCallback(MessageCallback& cb) { enqueueCallback = &cb; }
-void Message::resetEnqueueCompleteCallback() { enqueueCallback = 0; }
+void Message::setEnqueueCompleteCallback(MessageCallback& cb) {
+ sys::Mutex::ScopedLock l(callbackLock);
+ enqueueCallback = &cb;
+}
+
+void Message::resetEnqueueCompleteCallback() {
+ sys::Mutex::ScopedLock l(callbackLock);
+ enqueueCallback = 0;
+}
+
+void Message::setDequeueCompleteCallback(MessageCallback& cb) {
+ sys::Mutex::ScopedLock l(callbackLock);
+ dequeueCallback = &cb;
+}
-void Message::setDequeueCompleteCallback(MessageCallback& cb) { dequeueCallback = &cb; }
-void Message::resetDequeueCompleteCallback() { dequeueCallback = 0; }
+void Message::resetDequeueCompleteCallback() {
+ sys::Mutex::ScopedLock l(callbackLock);
+ dequeueCallback = 0;
+}
framing::FieldTable& Message::getOrInsertHeaders()
{