Thread safety fixes for race conditions on incoming messages.

* cpp/lib/client/MessageListener.h:  const correctness.
* cpp/tests/*: MessageListener const change.
* cpp/lib/broker/Content.h:  Removed out-of-date FIXME comments.
* cpp/lib/client/ClientChannel.h/	.cpp():
 - added locking for consumers map and other member access.
 - refactored implementations of Basic get, deliver, return: 
   most logic now encapsulted in IncomingMessage class.
 - fix channel close problems.
* cpp/lib/client/ClientMessage.h/.cpp:
 - const correctness & API convenience fixes.
 - getMethod/setMethod/getHeader: for new IncomingMessage
* cpp/lib/client/Connection.h/.cpp:
 - Fixes to channel closure.
* cpp/lib/client/IncomingMessage.h/.cpp:
 - Encapsulate *all* incoming message handling for client.
 - Moved handling of BasicGetOk to IncomingMessage to fix race.
 - Thread safety fixes.
* cpp/lib/client/ResponseHandler.h/.cpp:
 - added getResponse for ClientChannel.
* cpp/lib/common/Exception.h:
 - added missing throwSelf implementations.
 - added ShutdownException as general purpose shut-down indicator.
 - added EmptyException as general purpose "empty" indicator.
* cpp/lib/common/sys/Condition|Monitor|Mutex.h|.cpp:
 - Condition variable abstraction extracted from Monitor for situations
   where a single lock is associated with multiple conditions.
* cpp/tests/ClientChannelTest.cpp:
 - Test incoming message transfer, get, consume etc.


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-9@510161 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 9 insertions, 6 deletions

diff --git a/cpp/lib/client/Connection.cpp b/cpp/lib/client/Connection.cpp
index 5b97ca8e5d..566c8fc573 100644
--- a/cpp/lib/client/Connection.cpp
+++ b/cpp/lib/client/Connection.cpp
@@ -18,7 +18,9 @@
  * under the License.
  *
  */
+#include <algorithm>
 #include <boost/format.hpp>
+#include <boost/bind.hpp>
 
 #include <Connection.h>
 #include <ClientChannel.h>
@@ -27,7 +29,6 @@
 #include <iostream>
 #include <sstream>
 #include <MethodBodyInstances.h>
-#include <boost/bind.hpp>
 #include <functional>
 
 using namespace qpid::framing;
@@ -83,15 +84,17 @@ void Connection::close(
 {
     if(isOpen) {
         // TODO aconway 2007-01-29: Exception handling - could end up
-        // partly closed.
+        // partly closed with threads left unjoined.
         isOpen = false;
         channel0.sendAndReceive<ConnectionCloseOkBody>(
             new ConnectionCloseBody(
                 getVersion(), code, msg, classId, methodId));
-        while(!channels.empty()) {
-            channels.begin()->second->close();
-            channels.erase(channels.begin());
-        }
+
+        using boost::bind;
+        for_each(channels.begin(), channels.end(),
+                 bind(&Channel::closeInternal,
+                             bind(&ChannelMap::value_type::second, _1)));
+        channels.clear();
         connector->close();
     }
 }