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, 6 insertions, 15 deletions

diff --git a/cpp/lib/client/ClientChannel.h b/cpp/lib/client/ClientChannel.h
index ed67fd8f6b..9c422305b0 100644
--- a/cpp/lib/client/ClientChannel.h
+++ b/cpp/lib/client/ClientChannel.h
@@ -89,19 +89,12 @@ class Channel : public framing::ChannelAdapter,
         u_int64_t lastDeliveryTag;
     };
     typedef std::map<std::string, Consumer> ConsumerMap;
-    typedef std::queue<boost::shared_ptr<framing::AMQMethodBody> > IncomingMethods;
     
-    static const std::string OK;
-        
+    sys::Mutex lock;
     Connection* connection;
     sys::Thread dispatcher;
-    IncomingMethods incomingMethods;
-    IncomingMessage* incoming;
+    IncomingMessage incoming;
     ResponseHandler responses;
-    std::queue<IncomingMessage*> messages;//holds returned messages or those delivered for a consume
-    IncomingMessage* retrieved;//holds response to basic.get
-    sys::Monitor dispatchMonitor;
-    sys::Monitor retrievalMonitor;
     ConsumerMap consumers;
     ReturnedMessageHandler* returnsHandler;
 
@@ -109,10 +102,7 @@ class Channel : public framing::ChannelAdapter,
     const bool transactional;
     framing::ProtocolVersion version;
 
-    void enqueue();
     void retrieve(Message& msg);
-    IncomingMessage* dequeue();
-    void dispatch();
     void deliver(Consumer& consumer, Message& msg);
         
     void handleHeader(framing::AMQHeaderBody::shared_ptr body);
@@ -307,7 +297,8 @@ class Channel : public framing::ChannelAdapter,
      * receive this message on publication, the message will be
      * returned (see setReturnedMessageHandler()).
      */
-    void publish(Message& msg, const Exchange& exchange, const std::string& routingKey, 
+    void publish(const Message& msg, const Exchange& exchange,
+                 const std::string& routingKey, 
                  bool mandatory = false, bool immediate = false);
 
     /**
@@ -352,8 +343,8 @@ class Channel : public framing::ChannelAdapter,
      * Closing a channel that is not open has no effect.
      */
     void close(
-        framing::ReplyCode = 200, const std::string& =OK,
-        framing::ClassId = 0, framing::MethodId  = 0);
+        framing::ReplyCode = 200, const std::string& ="OK",
+         framing::ClassId = 0, framing::MethodId  = 0);
 
     /**
      * Set a handler for this channel that will process any