Ensure that callbacks are removed before IncompleteMessageList instance is deleted.

(also removed commented out line of code from Queue.cpp)



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

5 files changed, 11 insertions, 1 deletions

diff --git a/qpid/cpp/src/qpid/broker/IncompleteMessageList.cpp b/qpid/cpp/src/qpid/broker/IncompleteMessageList.cpp
index bfa7b5dd62..26c7a83d2f 100644
--- a/qpid/cpp/src/qpid/broker/IncompleteMessageList.cpp
+++ b/qpid/cpp/src/qpid/broker/IncompleteMessageList.cpp
@@ -28,6 +28,12 @@ IncompleteMessageList::IncompleteMessageList() :
     callback(boost::bind(&IncompleteMessageList::enqueueComplete, this, _1))
 {}
 
+IncompleteMessageList::~IncompleteMessageList() 
+{
+    sys::Mutex::ScopedLock l(lock);
+    std::for_each(incomplete.begin(), incomplete.end(), boost::bind(&Message::resetEnqueueCompleteCallback, _1));
+}
+
 void IncompleteMessageList::add(boost::intrusive_ptr<Message> msg)
 {
     sys::Mutex::ScopedLock l(lock);
diff --git a/qpid/cpp/src/qpid/broker/IncompleteMessageList.h b/qpid/cpp/src/qpid/broker/IncompleteMessageList.h
index 40c47cfaa6..f89c0023b0 100644
--- a/qpid/cpp/src/qpid/broker/IncompleteMessageList.h
+++ b/qpid/cpp/src/qpid/broker/IncompleteMessageList.h
@@ -44,6 +44,7 @@ public:
     typedef Message::MessageCallback CompletionListener;    
 
     IncompleteMessageList();
+    ~IncompleteMessageList();
     
     void add(boost::intrusive_ptr<Message> msg);
     void process(const CompletionListener& l, bool sync);
diff --git a/qpid/cpp/src/qpid/broker/Message.cpp b/qpid/cpp/src/qpid/broker/Message.cpp
index a99a10180e..c0a15d795c 100644
--- a/qpid/cpp/src/qpid/broker/Message.cpp
+++ b/qpid/cpp/src/qpid/broker/Message.cpp
@@ -361,7 +361,9 @@ void Message::allDequeuesComplete() {
 }
 
 void Message::setEnqueueCompleteCallback(MessageCallback& cb) { enqueueCallback = &cb; }
+void Message::resetEnqueueCompleteCallback() { enqueueCallback = 0; }
 
 void Message::setDequeueCompleteCallback(MessageCallback& cb) { dequeueCallback = &cb; }
+void Message::resetDequeueCompleteCallback() { dequeueCallback = 0; }
 
 }} // namespace qpid::broker
diff --git a/qpid/cpp/src/qpid/broker/Message.h b/qpid/cpp/src/qpid/broker/Message.h
index 8510ef78e9..bed191fb8d 100644
--- a/qpid/cpp/src/qpid/broker/Message.h
+++ b/qpid/cpp/src/qpid/broker/Message.h
@@ -146,9 +146,11 @@ public:
 
     /** Call cb when enqueue is complete, may call immediately. Holds cb by reference. */
     void setEnqueueCompleteCallback(MessageCallback& cb);
+    void resetEnqueueCompleteCallback();
 
     /** Call cb when dequeue is complete, may call immediately. Holds cb by reference. */
     void setDequeueCompleteCallback(MessageCallback& cb);
+    void resetDequeueCompleteCallback();
 
   private:
     typedef std::map<const Queue*,boost::intrusive_ptr<Message> > Replacement;
diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp
index c1cbceccb5..9f722cf8be 100644
--- a/qpid/cpp/src/qpid/broker/Queue.cpp
+++ b/qpid/cpp/src/qpid/broker/Queue.cpp
@@ -496,7 +496,6 @@ void Queue::push(boost::intrusive_ptr<Message>& msg){
         QueuedMessage qm(this, msg, ++sequence);
         if (policy.get()) policy->tryEnqueue(qm);
          
-        //if (lastValueQueue && LVQinsert(qm) ) return; // LVQ update of existing message
         LVQ::iterator i;
         if (lastValueQueue){
             const framing::FieldTable* ft = msg->getApplicationHeaders();