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

5 files changed, 182 insertions, 95 deletions

diff --git a/cpp/src/tests/BrokerFixture.h b/cpp/src/tests/BrokerFixture.h
new file mode 100644
index 0000000000..ceb8ecb4e6
--- /dev/null
+++ b/cpp/src/tests/BrokerFixture.h
@@ -0,0 +1,101 @@
+#ifndef TESTS_BROKERFIXTURE_H
+#define TESTS_BROKERFIXTURE_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 "qpid/sys/Thread.h"
+#include "qpid/broker/Broker.h"
+#include "qpid/client/Connection.h"
+#include "qpid/client/ConnectionImpl.h"
+#include "qpid/client/Session_0_10.h"
+#include "qpid/client/SubscriptionManager.h"
+
+namespace qpid { namespace client {
+/** Back door into private Connector stuff */
+struct TestConnector {
+    static void disconnect(qpid::client::Connector& c) {
+        c.socket.close();
+        c.handleClosed();
+    }
+};
+}}
+
+/**
+ * A fixture to create an in-process broker and connect to it for tests.
+ */
+struct BrokerFixture {
+    typedef qpid::broker::Broker Broker;
+    typedef boost::shared_ptr<Broker> BrokerPtr;
+
+    struct OpenConnection : public qpid::client::Connection {
+        OpenConnection(int port) { open("localhost", port); }
+    };
+    
+    BrokerPtr broker;
+    qpid::sys::Thread brokerThread;
+    OpenConnection connection;
+    qpid::client::Session_0_10 session;
+    qpid::client::SubscriptionManager subs;
+    qpid::client::LocalQueue lq;
+        
+    BrokerPtr newBroker() {
+        Broker::Options opts;
+        opts.port=0;
+        opts.workerThreads=1;
+        BrokerPtr b=Broker::create(opts);
+        // TODO aconway 2007-12-05: Without the following line
+        // the test can hang in the connection ctor. This is
+        // a race condition that should be fixed.
+        b->getPort(); 
+        return b;
+    };
+
+    BrokerFixture() : broker(newBroker()),
+                      brokerThread(*broker),
+                      connection(broker->getPort()),
+                      session(connection.newSession()),
+                      subs(session)
+    {}
+
+    ~BrokerFixture() {
+        connection.close();
+        broker->shutdown();
+        brokerThread.join();
+    }
+
+    /** Open a connection to the local broker */
+    void open(qpid::client::Connection& c) {
+        c.open("localhost", broker->getPort());
+    }
+
+    /** Close a connection's socket */ 
+    static void disconnect(qpid::client::Connection& c) {
+        struct Expose : public qpid::client::Connection {
+            void disconnect() {
+                qpid::client::TestConnector::disconnect(*impl->getConnector());
+            }
+        };
+        static_cast<Expose&>(c).disconnect();
+    }
+};
+
+#endif  /*!TESTS_BROKERFIXTURE_H*/
diff --git a/cpp/src/tests/ClientChannelTest.cpp b/cpp/src/tests/ClientChannelTest.cpp
index 454b9ca56d..bd8f5af6be 100644
--- a/cpp/src/tests/ClientChannelTest.cpp
+++ b/cpp/src/tests/ClientChannelTest.cpp
@@ -20,7 +20,7 @@
  */
 #include <vector>
 #include "qpid_test_plugin.h"
-#include "InProcessBroker.h"
+#include "BrokerFixture.h"
 #include "qpid/client/Channel.h"
 #include "qpid/client/Message.h"
 #include "qpid/client/Queue.h"
@@ -44,7 +44,7 @@ const size_t FRAME_MAX = 256;
  * The test base defines the tests methods, derived classes
  * instantiate the channel in Basic or Message mode.
  */
-class ChannelTestBase : public CppUnit::TestCase  
+class ChannelTestBase : public CppUnit::TestCase, public BrokerFixture
 {
     struct Listener: public qpid::client::MessageListener {
         vector<Message> messages;
@@ -56,7 +56,6 @@ class ChannelTestBase : public CppUnit::TestCase
         }
     };
     
-    qpid::InProcessBrokerClient connection; 
     const std::string qname;
     const std::string data;
     Queue queue;
@@ -69,8 +68,7 @@ class ChannelTestBase : public CppUnit::TestCase
   public:
 
     ChannelTestBase()
-        : connection(FRAME_MAX),
-          qname("testq"), data("hello"),
+        : qname("testq"), data("hello"),
           queue(qname, true), exchange("", Exchange::DIRECT_EXCHANGE)
     {}
 
diff --git a/cpp/src/tests/ClientSessionTest.cpp b/cpp/src/tests/ClientSessionTest.cpp
index ac2cf155f4..70e8a41074 100644
--- a/cpp/src/tests/ClientSessionTest.cpp
+++ b/cpp/src/tests/ClientSessionTest.cpp
@@ -19,7 +19,7 @@
  *
  */
 #include "qpid_test_plugin.h"
-#include "InProcessBroker.h"
+#include "BrokerFixture.h"
 #include "qpid/client/Dispatcher.h"
 #include "qpid/client/Session_0_10.h"
 #include "qpid/framing/TransferContent.h"
@@ -61,7 +61,7 @@ struct DummyListener : public MessageListener
     }
 };
 
-class ClientSessionTest : public CppUnit::TestCase
+class ClientSessionTest : public CppUnit::TestCase, public BrokerFixture
 {
     CPPUNIT_TEST_SUITE(ClientSessionTest);
     CPPUNIT_TEST(testQueueQuery);
@@ -74,23 +74,14 @@ class ClientSessionTest : public CppUnit::TestCase
     CPPUNIT_TEST_SUITE_END();
 
     shared_ptr<broker::Broker> broker;
-    Session_0_10 session;
-    // Defer construction & thread creation to setUp
-    boost::optional<InProcessConnection> c;
-    boost::optional<InProcessConnection> c2;
 
-public:
+  public:
 
     void setUp() {
         broker = broker::Broker::create();
-        c=boost::in_place<InProcessConnection>(broker);
-        c2=boost::in_place<InProcessConnection>(broker);
     }
 
     void tearDown() {
-        c2.reset();
-        c.reset();
-        broker.reset();
     }
 
     void declareSubscribe(const std::string& q="my-queue",
@@ -109,7 +100,7 @@ public:
     
     void testQueueQuery() 
     {
-        session = c->newSession();
+        session =connection.newSession();
         session.queueDeclare(queue="my-queue", alternateExchange="amq.fanout", exclusive=true, autoDelete=true);
         TypedResult<QueueQueryResult> result = session.queueQuery(std::string("my-queue"));
         CPPUNIT_ASSERT_EQUAL(false, result.get().getDurable());
@@ -120,7 +111,7 @@ public:
 
     void testTransfer()
     {
-        session = c->newSession();
+        session =connection.newSession();
         declareSubscribe();
         session.messageTransfer(content=TransferContent("my-message", "my-queue"));
         //get & test the message:
@@ -128,12 +119,12 @@ public:
         CPPUNIT_ASSERT(msg->isA<MessageTransferBody>());
         CPPUNIT_ASSERT_EQUAL(std::string("my-message"), msg->getContent());
         //confirm receipt:
-        session.execution().completed(msg->getId(), true, true);
+        session.getExecution().completed(msg->getId(), true, true);
     }
 
     void testDispatcher()
     {
-        session = c->newSession();
+        session =connection.newSession();
         declareSubscribe();
 
         TransferContent msg1("One");
@@ -161,16 +152,16 @@ public:
     }
 
     void testResumeExpiredError() {
-        session = c->newSession(0);
+        session =connection.newSession(0);
         session.suspend();  // session has 0 timeout.
         try {
-            c->resume(session);
+           connection.resume(session);
             CPPUNIT_FAIL("Expected InvalidArgumentException.");
         } catch(const InternalErrorException&) {}
     }
 
     void testUseSuspendedError() {
-        session = c->newSession(60);
+        session =connection.newSession(60);
         session.suspend();
         try {
             session.exchangeQuery(name="amq.fanout");
@@ -179,26 +170,27 @@ public:
     }
 
     void testSuspendResume() {
-        session = c->newSession(60);
+        session =connection.newSession(60);
         declareSubscribe();
         session.suspend();
         // Make sure we are still subscribed after resume.
-        c->resume(session);
+       connection.resume(session);
         session.messageTransfer(content=TransferContent("my-message", "my-queue"));
         FrameSet::shared_ptr msg = session.get();
         CPPUNIT_ASSERT_EQUAL(string("my-message"), msg->getContent());
     }
 
     void testDisconnectResume() {
-        session = c->newSession(60);
+        session =connection.newSession(60);
         session.queueDeclare(queue="before");
         CPPUNIT_ASSERT(queueExists("before"));
-        // Simulate lost frames.
-        c->discard();
         session.queueDeclare(queue=string("after"));
-        c->disconnect(); // Simulate disconnect, resume on a new connection.
-        c2->resume(session);
+        disconnect(connection);
+        Connection c2;
+        open(c2);
+        c2.resume(session);
         CPPUNIT_ASSERT(queueExists("after"));
+        c2.close();
     }
 };
 
diff --git a/cpp/src/tests/exception_test.cpp b/cpp/src/tests/exception_test.cpp
index d7094c9bfd..e774c48070 100644
--- a/cpp/src/tests/exception_test.cpp
+++ b/cpp/src/tests/exception_test.cpp
@@ -20,7 +20,7 @@
  */
 
 #include "unit_test.h"
-#include "InProcessBroker.h"
+#include "BrokerFixture.h"
 #include "qpid/client/SubscriptionManager.h"
 #include "qpid/sys/Runnable.h"
 #include "qpid/sys/Thread.h"
@@ -31,74 +31,74 @@ QPID_AUTO_TEST_SUITE(exception_test)
 
 using namespace std;
 using namespace qpid;
+using namespace sys;
 using namespace client;
 using namespace framing;
 
-struct Fixture {
-    InProcessConnection connection;
-    InProcessConnection connection2;
-    Session_0_10 session;
-    SubscriptionManager sub;
-    LocalQueue q;
+using boost::bind;
+using boost::function;
 
-    Fixture() : connection(),
-                connection2(connection.getBroker()),
-                session(connection.newSession()),
-                sub(session)
-    {
-        session.queueDeclare(arg::queue="q");
+template <class Ex>
+struct Catcher : public Runnable {
+    function<void ()> f;
+    bool caught;
+    Thread thread;
+    
+    Catcher(function<void ()> f_) : f(f_), caught(false), thread(this) {}
+    ~Catcher() { join(); }
+    
+    void run() {
+        try { f(); }
+        catch(const Ex& e) {
+            caught=true;
+            BOOST_MESSAGE(e.what());
+        }
+        catch(const std::exception& e) {
+            BOOST_ERROR(string("Bad exception: ")+e.what());
+        }
+        catch(...) {
+            BOOST_ERROR(string("Bad exception: unknown"));
+        }
+    }
+
+    bool join() {
+        if (thread.id()) {
+            thread.join();
+            thread=Thread();
+        }
+        return caught;
     }
 };
 
+BOOST_FIXTURE_TEST_CASE(DisconnectedGet, BrokerFixture) {
+    Catcher<ClosedException> get(bind(&Session_0_10::get, session));
+    disconnect(connection);
+    BOOST_CHECK(get.join());
+}
 
-// TODO aconway 2007-11-30: need InProcessBroker to be a more accurate
-// simulation of shutdown behaviour. It should override only 
-// Connector.run() to substitute NetworkQueues for the Dispatcher.
-// 
-// template <class Ex>
-// struct Catcher : public sys::Runnable {
-//     Session_0_10 s;
-//     boost::function<void ()> f;
-//     bool caught;
-//     Catcher(Session_0_10 s_, boost::function<void ()> f_)
-//         : s(s_), f(f_), caught(false) {}
-//     void run() {
-//         try { f(); } catch(const Ex& e) {
-//             caught=true;
-//             BOOST_MESSAGE(e.what());
-//         }
-//     }
-// };
+BOOST_FIXTURE_TEST_CASE(DisconnectedPop, BrokerFixture) {
+    session.queueDeclare(arg::queue="q");
+    subs.subscribe(lq, "q");
+    Catcher<ClosedException> pop(bind(&LocalQueue::pop, boost::ref(lq)));
+    disconnect(connection);
+    BOOST_CHECK(pop.join());
+}
 
-// BOOST_FIXTURE_TEST_CASE(DisconnectedGet, Fixture) {
-//     Catcher<Exception> get(session, boost::bind(&Session_0_10::get, session));
-//     sub.subscribe(q, "q");
-//     sys::Thread t(get);
-//     connection.disconnect();
-//     t.join();
-//     BOOST_CHECK(get.caught);
-// }
-
-// BOOST_FIXTURE_TEST_CASE(DisconnectedListen, Fixture) {
+// FIXME aconway 2007-12-07: This test hangs sporadically at t.join
+// BOOST_FIXTURE_TEST_CASE(DisconnectedListen, BrokerFixture) {
 //     struct NullListener : public MessageListener {
 //         void received(Message&) { BOOST_FAIL("Unexpected message"); }
 //     } l;
-//     sub.subscribe(l, "q");
-//     connection.disconnect();
-//     try {
-//         sub.run();
-//         BOOST_FAIL("Expected exception");
-//     } catch (const Exception&e) { BOOST_FAIL(e.what()); }
-//     try {
-//         session.queueDeclare(arg::queue="foo");
-//         BOOST_FAIL("Expected exception");
-//     } catch (const Exception&e) { BOOST_FAIL(e.what()); }
+//     session.queueDeclare(arg::queue="q");
+//     subs.subscribe(l, "q");
+//     Thread t(subs);
+//     disconnect(connection);
+//     t.join();
+//     BOOST_CHECK_THROW(session.close(), InternalErrorException);    
 // }
 
-// TODO aconway 2007-11-30: setSynchronous appears not to work.
-// BOOST_FIXTURE_TEST_CASE(NoSuchQueueTest, Fixture) {
-//     session.setSynchronous(true);
-//     BOOST_CHECK_THROW(sub.subscribe(q, "no such queue"), NotFoundException);
-// }
+BOOST_FIXTURE_TEST_CASE(NoSuchQueueTest, BrokerFixture) {
+    BOOST_CHECK_THROW(subs.subscribe(lq, "no such queue").sync(), NotFoundException);
+}
 
 QPID_AUTO_TEST_SUITE_END()
diff --git a/cpp/src/tests/perftest.cpp b/cpp/src/tests/perftest.cpp
index bd2c0d9933..d6185a0100 100644
--- a/cpp/src/tests/perftest.cpp
+++ b/cpp/src/tests/perftest.cpp
@@ -205,7 +205,7 @@ struct Setup : public Client {
             }
         }
         // Make sure this is all completed before we return.
-        session.execution().sendSyncRequest();
+        session.getExecution().sendSyncRequest();
     }
 };
 
@@ -231,13 +231,9 @@ class Stats {
     
     // Functor to collect rates.
     void operator()(const string& data) {
-        try {
-            double d=lexical_cast<double>(data);
-            values.push_back(d);
-            sum += d;
-        } catch (...) {
-            throw Exception(QPID_MSG("Bad data, expecting double: " << data));
-        }
+        double d=lexical_cast<double>(data);
+        values.push_back(d);
+        sum += d;
     }
     
     double mean() const {