Refactored client::Message to be independent of all Basic class concepts

and client::IncomingMessage to handle 0-9 style references and appends.

* cpp/lib/client/ClientMessage.cpp: Made independent of Basic class.
* cpp/lib/client/IncomingMessage.cpp: Refactored to handle references/appends.
* cpp/lib/client/BasicMessageChannel.cpp: Refactored to use new IncomingMessage

Thread safety fixes:
* cpp/lib/client/ResponseHandler.h: Remove stateful functions.
* cpp/lib/client/ClientChannel.cpp: use new ResponseHandler interface.

Minor cleanup:
* cpp/lib/common/framing/BasicHeaderProperties.cpp: use DeliveryMode enum.
* cpp/tests/HeaderTest.cpp:  use DeliveryMode enum.
* cpp/tests/MessageTest.cpp:  use DeliveryMode enum.
* cpp/lib/common/shared_ptr.h: #include <boost/cast.hpp> for convenience.
* cpp/lib/common/sys/ThreadSafeQueue.h: Changed "stop"  "shutdown" 
* cpp/lib/common/sys/ProducerConsumer.h: Changed "stop"  "shutdown"  
* cpp/tests/ClientChannelTest.cpp (TestCase): Removed debug couts.
* cpp/tests/setup: valgrind --demangle=yes by default.
* cpp/tests/topictest: sleep to hack around startup race.
* cpp/lib/broker/BrokerQueue.cpp (configure): Fixed memory leak.

Removed/updated FIXME comments in: 
* cpp/lib/broker/BrokerMessage.cpp: 
* cpp/lib/broker/BrokerMessageBase.h: 
* cpp/lib/broker/InMemoryContent.cpp: 
* cpp/lib/common/framing/MethodContext.h: 



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

1 files changed, 25 insertions, 32 deletions

diff --git a/cpp/lib/client/ResponseHandler.cpp b/cpp/lib/client/ResponseHandler.cpp
index 4498de41ae..926a9ce336 100644
--- a/cpp/lib/client/ResponseHandler.cpp
+++ b/cpp/lib/client/ResponseHandler.cpp
@@ -18,12 +18,10 @@
  * under the License.
  *
  */
-#include <boost/format.hpp>
-
-#include <ResponseHandler.h>
-#include <sys/Monitor.h>
 #include <QpidError.h>
-#include "amqp_types.h"
+#include <boost/format.hpp>
+#include "ResponseHandler.h"
+#include "AMQMethodBody.h"
 
 using namespace qpid::sys;
 using namespace qpid::framing;
@@ -31,56 +29,51 @@ using namespace qpid::framing;
 namespace qpid {
 namespace client {
 
-ResponseHandler::ResponseHandler() : waiting(false){}
+ResponseHandler::ResponseHandler() : waiting(false), shutdownFlag(false) {}
 
 ResponseHandler::~ResponseHandler(){}
 
-bool ResponseHandler::validate(ClassId c, MethodId  m) {
-    return response != 0 &&
-        response->amqpClassId() ==c && response->amqpMethodId() == m;
+bool ResponseHandler::isWaiting()  {
+    Monitor::ScopedLock l(monitor);
+    return waiting;
 }
 
-void ResponseHandler::waitForResponse(){
+void ResponseHandler::expect(){
     Monitor::ScopedLock l(monitor);
-    while (waiting)
-	monitor.wait();
+    waiting = true;
 }
 
-void ResponseHandler::signalResponse(
-    qpid::framing::AMQMethodBody::shared_ptr _response)
+void ResponseHandler::signalResponse(MethodPtr _response)
 {
     Monitor::ScopedLock l(monitor);
     response = _response;
+    if (!response)
+        shutdownFlag=true;
     waiting = false;
     monitor.notify();
 }
 
-void ResponseHandler::receive(ClassId c, MethodId  m) {
+ResponseHandler::MethodPtr ResponseHandler::receive() {
     Monitor::ScopedLock l(monitor);
-    while (waiting)
+    while (!response && !shutdownFlag)
 	monitor.wait();
-    getResponse(); // Check for closed.
-    if(!validate(response->amqpClassId(), response->amqpMethodId())) {
+    if (shutdownFlag) 
+        THROW_QPID_ERROR(
+            PROTOCOL_ERROR, "Channel closed unexpectedly.");
+    MethodPtr result = response;
+    response.reset();
+    return result;
+}
+
+ResponseHandler::MethodPtr ResponseHandler::receive(ClassId c, MethodId  m) {
+    MethodPtr response = receive();
+    if(c != response->amqpClassId() || m != response->amqpMethodId()) {
 	THROW_QPID_ERROR(
             PROTOCOL_ERROR,
             boost::format("Expected class:method %d:%d, got %d:%d")
             % c % m % response->amqpClassId() % response->amqpMethodId());
     }
-}
-
-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();
-}
-void ResponseHandler::expect(){
-    waiting = true;
-}
-
 }} // namespace qpid::client