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, 5 deletions

diff --git a/cpp/lib/client/ResponseHandler.cpp b/cpp/lib/client/ResponseHandler.cpp
index ea48fa2386..4498de41ae 100644
--- a/cpp/lib/client/ResponseHandler.cpp
+++ b/cpp/lib/client/ResponseHandler.cpp
@@ -59,11 +59,8 @@ void ResponseHandler::receive(ClassId c, MethodId  m) {
     Monitor::ScopedLock l(monitor);
     while (waiting)
 	monitor.wait();
-    if (!response) {
-        THROW_QPID_ERROR(
-            PROTOCOL_ERROR, "Channel closed unexpectedly.");
-    }
-     if(!validate(response->amqpClassId(), response->amqpMethodId())) {
+    getResponse(); // Check for closed.
+    if(!validate(response->amqpClassId(), response->amqpMethodId())) {
 	THROW_QPID_ERROR(
             PROTOCOL_ERROR,
             boost::format("Expected class:method %d:%d, got %d:%d")
@@ -71,6 +68,13 @@ void ResponseHandler::receive(ClassId c, MethodId  m) {
     }
 }
 
+framing::AMQMethodBody::shared_ptr ResponseHandler::getResponse() {
+    if (!response) 
+        THROW_QPID_ERROR(
+            PROTOCOL_ERROR, "Channel closed unexpectedly.");
+    return response;
+}
+
 RequestId ResponseHandler::getRequestId() {
     assert(response->getRequestId());
     return response->getRequestId();