Summary:

 - Replaced InProcessBroker with BrokerFixture, uses a full loopback
   broker for more realistic tests.
 - Extracted non-generated parts of Session_0_10 into SessionBase.
 - Sundry small fixes.

src/tests/BrokerFixture.h
 - in process broker with loopback connections.
 - tests can force a disorderly disconnect.

src/qpid/client/Connector.h
 - back door to private members for BrokerFixture.
 - close() in destructor to avoid leaks.

src/qpid/client/ConnectionImpl.h,cpp:
 - close() in destructor, to fix hang when destroyed without being closed.

src/qpid/client/CompletionTracker.h,.cpp:
 - Fixed race in close/add.

src/qpid/client/SessionBase.h,cpp:
 - Extracted all non-generated code from Session_0_10 into SessionBase
 - Added sync()
 
src/tests/exception_test.cpp: Converted to boost & BrokerFixture

src/tests/ClientChannelTest.cpp, ClientSessionTest.cpp: Use BrokerFixture


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

1 files changed, 19 insertions, 11 deletions

diff --git a/cpp/src/qpid/client/SubscriptionManager.h b/cpp/src/qpid/client/SubscriptionManager.h
index c9066fe1de..1205478bf8 100644
--- a/cpp/src/qpid/client/SubscriptionManager.h
+++ b/cpp/src/qpid/client/SubscriptionManager.h
@@ -23,21 +23,24 @@
  */
 #include "qpid/sys/Mutex.h"
 #include <qpid/client/Dispatcher.h>
+#include <qpid/client/Completion.h>
 #include <qpid/client/Session_0_10.h>
 #include <qpid/client/MessageListener.h>
 #include <qpid/client/LocalQueue.h>
+#include <qpid/sys/Runnable.h>
+
 #include <set>
 #include <sstream>
 
 namespace qpid {
 namespace client {
 
-class SubscriptionManager
+class SubscriptionManager : public sys::Runnable
 {
     typedef sys::Mutex::ScopedLock Lock;
     typedef sys::Mutex::ScopedUnlock Unlock;
 
-    void subscribeInternal(const std::string& q, const std::string& dest);
+    Completion subscribeInternal(const std::string& q, const std::string& dest);
     
     qpid::client::Dispatcher dispatcher;
     qpid::client::Session_0_10& session;
@@ -47,8 +50,9 @@ class SubscriptionManager
     AckPolicy autoAck;
     bool confirmMode;
     bool acquireMode;
-
-public:
+    bool autoStop;
+    
+  public:
     SubscriptionManager(Session_0_10& session);
     
     /**
@@ -59,9 +63,9 @@ public:
      *@param tag Unique destination tag for the listener.
      * If not specified, the queue name is used.
      */
-    void subscribe(MessageListener& listener,
-                   const std::string& queue,
-                   const std::string& tag=std::string());
+    Completion subscribe(MessageListener& listener,
+                         const std::string& queue,
+                         const std::string& tag=std::string());
 
     /**
      * Subscribe a LocalQueue to receive messages from queue.
@@ -70,17 +74,21 @@ public:
      *@param tag Unique destination tag for the listener.
      * If not specified, the queue name is used.
      */
-    void subscribe(LocalQueue& localQueue,
+    Completion subscribe(LocalQueue& localQueue,
                    const std::string& queue,
                    const std::string& tag=std::string());
 
     /** Cancel a subscription. */
     void cancel(const std::string tag);
 
-    /** Deliver messages until stop() is called.
-     *@param autoStop If true, return when all listeners are cancelled.
+    /** Deliver messages until stop() is called. */
+    void run();
+
+    /** If set true, run() will stop when all subscriptions
+     * are cancelled. If false, run will only stop when stop()
+     * is called. True by default.
      */
-    void run(bool autoStop=true);
+    void setAutoStop(bool set=true);
 
     /** Cause run() to return */
     void stop();