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

7 files changed, 168 insertions, 83 deletions

diff --git a/cpp/lib/common/Exception.cpp b/cpp/lib/common/Exception.cpp
index 0161518011..f7d91498e0 100644
--- a/cpp/lib/common/Exception.cpp
+++ b/cpp/lib/common/Exception.cpp
@@ -39,4 +39,8 @@ Exception* Exception::clone() const throw() { return new Exception(*this); }
 
 void Exception::throwSelf() const  { throw *this; }
 
+ShutdownException::ShutdownException() : Exception("Shut down.") {}
+
+EmptyException::EmptyException() : Exception("Empty.") {}
+
 } // namespace qpid
diff --git a/cpp/lib/common/Exception.h b/cpp/lib/common/Exception.h
index a07e85e7e3..b8cd68cf8c 100644
--- a/cpp/lib/common/Exception.h
+++ b/cpp/lib/common/Exception.h
@@ -66,6 +66,7 @@ struct ChannelException : public Exception {
     template <class T>
     ChannelException(framing::ReplyCode code_, const T& message)
         : Exception(message), code(code_) {}
+    void throwSelf() const { throw *this; }
 };
 
 struct ConnectionException : public Exception {
@@ -73,8 +74,23 @@ struct ConnectionException : public Exception {
     template <class T>
     ConnectionException(framing::ReplyCode code_, const T& message)
         : Exception(message), code(code_) {}
+    void throwSelf() const { throw *this; }
 };
 
+/**
+ * Exception used to indicate that a thread should shut down.
+ * Does not indicate an error that should be signalled to the user.
+ */
+struct ShutdownException : public Exception {
+    ShutdownException();
+    void throwSelf() const { throw *this; }
+};
+
+/** Exception to indicate empty queue or other empty state */
+struct EmptyException : public Exception {
+    EmptyException();
+    void throwSelf() const { throw *this; }
+};
 
 }
 
diff --git a/cpp/lib/common/ExceptionHolder.h b/cpp/lib/common/ExceptionHolder.h
index 1e5a30e27c..2769455aba 100644
--- a/cpp/lib/common/ExceptionHolder.h
+++ b/cpp/lib/common/ExceptionHolder.h
@@ -19,11 +19,15 @@
  *
  */
 
+#include <assert.h>
 #include <Exception.h>
 #include <boost/shared_ptr.hpp>
 
 namespace qpid {
 
+// FIXME aconway 2007-02-20:  Not necessary, a simple
+// Exception::shared_ptr will do the job. Remove
+// 
 /**
  * Holder for a heap-allocated exc eption that can be stack allocated
  * and thrown safely.
@@ -49,11 +53,11 @@ class ExceptionHolder : public Exception, public boost::shared_ptr<Exception>
 
     ~ExceptionHolder() throw() {}
 
-    const char* what() const throw() { return (*this)->what(); }
-    std::string toString() const throw() { return (*this)->toString(); }
-    virtual Exception* clone() const throw() { return (*this)->clone(); }
-    virtual void throwSelf() const { (*this)->throwSelf(); }
-    virtual void throwIf() const { if (*this) (*this)->throwSelf(); }
+    const char* what() const throw() { return get()->what(); }
+    std::string toString() const throw() { return get()->toString(); }
+    Exception* clone() const throw() { return get()->clone(); }
+    void throwIf() const { if (get()) get()->throwSelf(); }
+    void throwSelf() const { assert(get()); get()->throwSelf(); }
 };
 
 } // namespace qpid
diff --git a/cpp/lib/common/framing/ChannelAdapter.h b/cpp/lib/common/framing/ChannelAdapter.h
index 36362a417a..f6e3986eed 100644
--- a/cpp/lib/common/framing/ChannelAdapter.h
+++ b/cpp/lib/common/framing/ChannelAdapter.h
@@ -34,6 +34,8 @@ namespace framing {
 
 class MethodContext;
 
+// FIXME aconway 2007-02-20: Rename as ChannelBase or just Channel.
+
 /**
  * Base class for client and broker channels.
  *
diff --git a/cpp/lib/common/sys/Condition.h b/cpp/lib/common/sys/Condition.h
new file mode 100644
index 0000000000..9d70af5b84
--- /dev/null
+++ b/cpp/lib/common/sys/Condition.h
@@ -0,0 +1,128 @@
+#ifndef _sys_Condition_h
+#define _sys_Condition_h
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <sys/errno.h>
+#include <boost/noncopyable.hpp>
+#include <sys/Mutex.h>
+#include <sys/Time.h>
+
+#ifdef USE_APR
+#  include <apr_thread_cond.h>
+#endif
+
+namespace qpid {
+namespace sys {
+
+/**
+ * A condition variable for thread synchronization.
+ */
+class Condition
+{
+  public:
+    inline Condition();
+    inline ~Condition();
+    inline void wait(Mutex&);
+    inline bool wait(Mutex&, const Time& absoluteTime);
+    inline void notify();
+    inline void notifyAll();
+
+  private:
+#ifdef USE_APR
+    apr_thread_cond_t* condition;
+#else
+    pthread_cond_t condition;
+#endif
+};
+
+
+// APR ================================================================
+#ifdef USE_APR
+
+Condition::Condition() {
+    CHECK_APR_SUCCESS(apr_thread_cond_create(&condition, APRPool::get()));
+}
+
+Condition::~Condition() {
+    CHECK_APR_SUCCESS(apr_thread_cond_destroy(condition));
+}
+
+void Condition::wait(Mutex& mutex) {
+    CHECK_APR_SUCCESS(apr_thread_cond_wait(condition, mutex.mutex));
+}
+
+bool Condition::wait(Mutex& mutex, const Time& absoluteTime){
+    // APR uses microseconds.
+    apr_status_t status =
+        apr_thread_cond_timedwait(
+            condition, mutex.mutex, absoluteTime/TIME_USEC);
+    if(status != APR_TIMEUP) CHECK_APR_SUCCESS(status);
+    return status == 0;
+}
+
+void Condition::notify(){
+    CHECK_APR_SUCCESS(apr_thread_cond_signal(condition));
+}
+
+void Condition::notifyAll(){
+    CHECK_APR_SUCCESS(apr_thread_cond_broadcast(condition));
+}
+
+#else
+// POSIX ================================================================
+
+Condition::Condition() {
+    QPID_POSIX_THROW_IF(pthread_cond_init(&condition, 0));
+}
+
+Condition::~Condition() {
+    QPID_POSIX_THROW_IF(pthread_cond_destroy(&condition));
+}
+
+void Condition::wait(Mutex& mutex) {
+    QPID_POSIX_THROW_IF(pthread_cond_wait(&condition, &mutex.mutex));
+}
+
+bool Condition::wait(Mutex& mutex, const Time& absoluteTime){
+    struct timespec ts;
+    toTimespec(ts, absoluteTime);
+    int status = pthread_cond_timedwait(&condition, &mutex.mutex, &ts);
+    if (status != 0) {
+        if (status == ETIMEDOUT) return false;
+        throw QPID_POSIX_ERROR(status);
+    }
+    return true;
+}
+
+void Condition::notify(){
+    QPID_POSIX_THROW_IF(pthread_cond_signal(&condition));
+}
+
+void Condition::notifyAll(){
+    QPID_POSIX_THROW_IF(pthread_cond_broadcast(&condition));
+}
+#endif  /*USE_APR*/
+
+
+}}
+#endif  /*!_sys_Condition_h*/
diff --git a/cpp/lib/common/sys/Monitor.h b/cpp/lib/common/sys/Monitor.h
index e58931e699..a3bbd3c5aa 100644
--- a/cpp/lib/common/sys/Monitor.h
+++ b/cpp/lib/common/sys/Monitor.h
@@ -23,9 +23,7 @@
  */
 
 #include <sys/errno.h>
-#include <boost/noncopyable.hpp>
-#include <sys/Mutex.h>
-#include <sys/Time.h>
+#include <sys/Condition.h>
 
 #ifdef USE_APR
 #  include <apr_thread_cond.h>
@@ -37,91 +35,21 @@ namespace sys {
 /**
  * A monitor is a condition variable and a mutex
  */
-class Monitor : public Mutex
-{
+class Monitor : public Mutex, public Condition {
   public:
-    inline Monitor();
-    inline ~Monitor();
+    using Condition::wait;
     inline void wait();
     inline bool wait(const Time& absoluteTime);
-    inline void notify();
-    inline void notifyAll();
-
-  private:
-#ifdef USE_APR
-    apr_thread_cond_t* condition;
-#else
-    pthread_cond_t condition;
-#endif
 };
 
 
-// APR ================================================================
-#ifdef USE_APR
-
-Monitor::Monitor() {
-    CHECK_APR_SUCCESS(apr_thread_cond_create(&condition, APRPool::get()));
-}
-
-Monitor::~Monitor() {
-    CHECK_APR_SUCCESS(apr_thread_cond_destroy(condition));
-}
-
-void Monitor::wait() {
-    CHECK_APR_SUCCESS(apr_thread_cond_wait(condition, mutex));
-}
-
-bool Monitor::wait(const Time& absoluteTime){
-    // APR uses microseconds.
-    apr_status_t status =
-        apr_thread_cond_timedwait(condition, mutex, absoluteTime/TIME_USEC);
-    if(status != APR_TIMEUP) CHECK_APR_SUCCESS(status);
-    return status == 0;
-}
-
-void Monitor::notify(){
-    CHECK_APR_SUCCESS(apr_thread_cond_signal(condition));
-}
-
-void Monitor::notifyAll(){
-    CHECK_APR_SUCCESS(apr_thread_cond_broadcast(condition));
-}
-
-#else
-// POSIX ================================================================
-
-Monitor::Monitor() {
-    QPID_POSIX_THROW_IF(pthread_cond_init(&condition, 0));
-}
-
-Monitor::~Monitor() {
-    QPID_POSIX_THROW_IF(pthread_cond_destroy(&condition));
-}
-
 void Monitor::wait() {
-    QPID_POSIX_THROW_IF(pthread_cond_wait(&condition, &mutex));
-}
-
-bool Monitor::wait(const Time& absoluteTime){
-    struct timespec ts;
-    toTimespec(ts, absoluteTime);
-    int status = pthread_cond_timedwait(&condition, &mutex, &ts);
-    if (status != 0) {
-        if (status == ETIMEDOUT) return false;
-        throw QPID_POSIX_ERROR(status);
-    }
-    return true;
+    Condition::wait(*this);
 }
 
-void Monitor::notify(){
-    QPID_POSIX_THROW_IF(pthread_cond_signal(&condition));
+bool Monitor::wait(const Time& absoluteTime) {
+    return Condition::wait(*this, absoluteTime);
 }
 
-void Monitor::notifyAll(){
-    QPID_POSIX_THROW_IF(pthread_cond_broadcast(&condition));
-}
-#endif  /*USE_APR*/
-
-
 }}
 #endif  /*!_sys_Monitor_h*/
diff --git a/cpp/lib/common/sys/Mutex.h b/cpp/lib/common/sys/Mutex.h
index 87d537fb9e..9db9be0981 100644
--- a/cpp/lib/common/sys/Mutex.h
+++ b/cpp/lib/common/sys/Mutex.h
@@ -32,6 +32,8 @@
 namespace qpid {
 namespace sys {
 
+class Condition;
+
 /**
  * Scoped lock template: calls lock() in ctor, unlock() in dtor.
  * L can be any class with lock() and unlock() functions.
@@ -76,6 +78,7 @@ class Mutex : private boost::noncopyable {
 #else
     pthread_mutex_t mutex;
 #endif
+  friend class Condition;
 };
 
 #ifdef USE_APR