Fix leak of Decoder map entries.

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

3 files changed, 10 insertions, 8 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 910699f913..677bd2b722 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -172,6 +172,7 @@ void Cluster::addShadowConnection(const boost::intrusive_ptr<Connection>& c) {
 // Called by Connection::deliverClose() in deliverFrameQueue thread.
 void Cluster::erase(const ConnectionId& id) {
     connections.erase(id);
+    decoder.erase(id);
 }
 
 std::vector<string> Cluster::getIds() const {
diff --git a/cpp/src/qpid/cluster/Decoder.cpp b/cpp/src/qpid/cluster/Decoder.cpp
index b337ef43f4..6f65b3852a 100644
--- a/cpp/src/qpid/cluster/Decoder.cpp
+++ b/cpp/src/qpid/cluster/Decoder.cpp
@@ -29,6 +29,7 @@ namespace qpid {
 namespace cluster {
 
 void Decoder::decode(const EventHeader& eh, const char* data) {
+    sys::Mutex::ScopedLock l(lock);
     assert(eh.getType() == DATA); // Only handle connection data events.
     const char* cp = static_cast<const char*>(data);
     framing::Buffer buf(const_cast<char*>(cp), eh.getSize());
@@ -36,26 +37,24 @@ void Decoder::decode(const EventHeader& eh, const char* data) {
     if (decoder.decode(buf)) {  // Decoded a frame
         framing::AMQFrame frame(decoder.getFrame());
         while (decoder.decode(buf)) {
-            process(EventFrame(eh, frame));
+            callback(EventFrame(eh, frame));
             frame = decoder.getFrame();
         }
         // Set read-credit on the last frame ending in this event.
         // Credit will be given when this frame is processed.
-        process(EventFrame(eh, frame, 1)); 
+        callback(EventFrame(eh, frame, 1)); 
     }
     else {
         // We must give 1 unit read credit per event.
         // This event does not complete any frames so 
         // send an empty frame with the read credit.
-        process(EventFrame(eh, framing::AMQFrame(), 1));
+        callback(EventFrame(eh, framing::AMQFrame(), 1));
     }    
 }
 
-void Decoder::process(const EventFrame& ef) {
-    //need to check that this is not the empty frame mentioned above
-    if (ef.frame.getBody() && ef.frame.getMethod() && ef.frame.getMethod()->isA<framing::ClusterConnectionDeliverCloseBody>())
-        map.erase(ef.connectionId);
-    callback(ef);
+void Decoder::erase(const ConnectionId& c) {
+    sys::Mutex::ScopedLock l(lock);
+    map.erase(c);
 }
 
 }} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/Decoder.h b/cpp/src/qpid/cluster/Decoder.h
index acde4258a2..3fb7b4f73d 100644
--- a/cpp/src/qpid/cluster/Decoder.h
+++ b/cpp/src/qpid/cluster/Decoder.h
@@ -24,6 +24,7 @@
 
 #include "types.h"
 #include "qpid/framing/FrameDecoder.h"
+#include "qpid/sys/Mutex.h"
 #include <boost/function.hpp>
 #include <map>
 
@@ -48,6 +49,7 @@ class Decoder
 
   private:
     typedef std::map<ConnectionId, framing::FrameDecoder> Map;
+    sys::Mutex lock;
     Map map;
     void process(const EventFrame&);
     FrameHandler callback;