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, 21 insertions, 16 deletions

diff --git a/cpp/lib/client/ClientMessage.cpp b/cpp/lib/client/ClientMessage.cpp
index 8b08f7e535..bd4adb78f7 100644
--- a/cpp/lib/client/ClientMessage.cpp
+++ b/cpp/lib/client/ClientMessage.cpp
@@ -19,7 +19,6 @@
  *
  */
 #include <ClientMessage.h>
-
 using namespace qpid::client;
 using namespace qpid::framing;
 
@@ -40,63 +39,63 @@ Message::Message(AMQHeaderBody::shared_ptr& _header) : header(_header){
 Message::~Message(){
 }
 	
-BasicHeaderProperties* Message::getHeaderProperties(){
+BasicHeaderProperties* Message::getHeaderProperties() const {
     return dynamic_cast<BasicHeaderProperties*>(header->getProperties());
 }
 
-const std::string& Message::getContentType(){ 
+const std::string& Message::getContentType() const { 
     return getHeaderProperties()->getContentType(); 
 }
 
-const std::string& Message::getContentEncoding(){ 
+const std::string& Message::getContentEncoding() const { 
     return getHeaderProperties()->getContentEncoding(); 
 }
 
-FieldTable& Message::getHeaders(){ 
+FieldTable& Message::getHeaders() const { 
     return getHeaderProperties()->getHeaders(); 
 }
 
-u_int8_t Message::getDeliveryMode(){ 
+u_int8_t Message::getDeliveryMode() const { 
     return getHeaderProperties()->getDeliveryMode(); 
 }
 
-u_int8_t Message::getPriority(){ 
+u_int8_t Message::getPriority() const { 
     return getHeaderProperties()->getPriority(); 
 }
 
-const std::string& Message::getCorrelationId(){
+const std::string& Message::getCorrelationId() const {
     return getHeaderProperties()->getCorrelationId(); 
 }
 
-const std::string& Message::getReplyTo(){ 
+const std::string& Message::getReplyTo() const { 
     return getHeaderProperties()->getReplyTo(); 
 }
 
-const std::string& Message::getExpiration(){ 
+const std::string& Message::getExpiration() const { 
     return getHeaderProperties()->getExpiration(); 
 }
 
-const std::string& Message::getMessageId(){
+const std::string& Message::getMessageId() const {
     return getHeaderProperties()->getMessageId(); 
 }
 
-u_int64_t Message::getTimestamp(){ 
+u_int64_t Message::getTimestamp() const { 
     return getHeaderProperties()->getTimestamp(); 
 }
 
-const std::string& Message::getType(){ 
+const std::string& Message::getType() const { 
     return getHeaderProperties()->getType(); 
 }
 
-const std::string& Message::getUserId(){ 
+const std::string& Message::getUserId() const { 
     return getHeaderProperties()->getUserId(); 
 }
 
-const std::string& Message::getAppId(){ 
+const std::string& Message::getAppId() const { 
     return getHeaderProperties()->getAppId(); 
 }
 
-const std::string& Message::getClusterId(){ 
+const std::string& Message::getClusterId() const { 
     return getHeaderProperties()->getClusterId(); 
 }
 
@@ -155,3 +154,9 @@ void Message::setAppId(const std::string& appId){
 void Message::setClusterId(const std::string& clusterId){ 
     getHeaderProperties()->setClusterId(clusterId); 
 }
+
+
+u_int64_t Message::getDeliveryTag() const {
+    BasicDeliverBody* deliver=dynamic_cast<BasicDeliverBody*>(method.get());
+    return deliver ? deliver->getDeliveryTag() : 0;
+}