src/tests/SocketProxy.h: proxy between local client & server to simulate network disconnect.

src/qpid/client/Connector.h: remove friend hack for previous flawed disconnect approach.
src/tests/BrokerFixture.h:   ""
src/tests/ClientSessionTest.cpp, exception_test.cpp: use ProxyConnection


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

6 files changed, 108 insertions, 43 deletions

diff --git a/qpid/cpp/src/qpid/client/Connector.h b/qpid/cpp/src/qpid/client/Connector.h
index 946289efdc..b1d759569c 100644
--- a/qpid/cpp/src/qpid/client/Connector.h
+++ b/qpid/cpp/src/qpid/client/Connector.h
@@ -88,7 +88,6 @@ class Connector : public framing::OutputHandler,
     void eof(qpid::sys::AsynchIO&);
     
   friend class Channel;
-  friend class TestConnector;
 
   public:
     Connector(framing::ProtocolVersion pVersion,
diff --git a/qpid/cpp/src/tests/BrokerFixture.h b/qpid/cpp/src/tests/BrokerFixture.h
index ceb8ecb4e6..c7222c03a2 100644
--- a/qpid/cpp/src/tests/BrokerFixture.h
+++ b/qpid/cpp/src/tests/BrokerFixture.h
@@ -29,16 +29,6 @@
 #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.
  */
@@ -86,16 +76,6 @@ struct BrokerFixture {
     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/qpid/cpp/src/tests/ClientSessionTest.cpp b/qpid/cpp/src/tests/ClientSessionTest.cpp
index 70e8a41074..a960cea230 100644
--- a/qpid/cpp/src/tests/ClientSessionTest.cpp
+++ b/qpid/cpp/src/tests/ClientSessionTest.cpp
@@ -20,6 +20,7 @@
  */
 #include "qpid_test_plugin.h"
 #include "BrokerFixture.h"
+#include "SocketProxy.h"
 #include "qpid/client/Dispatcher.h"
 #include "qpid/client/Session_0_10.h"
 #include "qpid/framing/TransferContent.h"
@@ -181,14 +182,15 @@ class ClientSessionTest : public CppUnit::TestCase, public BrokerFixture
     }
 
     void testDisconnectResume() {
-        session =connection.newSession(60);
-        session.queueDeclare(queue="before");
+        ProxyConnection c(broker->getPort());
+        Session_0_10 s = c.session;
+        s.queueDeclare(queue="before");
         CPPUNIT_ASSERT(queueExists("before"));
-        session.queueDeclare(queue=string("after"));
-        disconnect(connection);
+        s.queueDeclare(queue=string("after"));
+        c.proxy.client.close();       // Disconnect the client.
         Connection c2;
         open(c2);
-        c2.resume(session);
+        c2.resume(s);
         CPPUNIT_ASSERT(queueExists("after"));
         c2.close();
     }
diff --git a/qpid/cpp/src/tests/Makefile.am b/qpid/cpp/src/tests/Makefile.am
index dc20486a7e..1e9a3342c0 100644
--- a/qpid/cpp/src/tests/Makefile.am
+++ b/qpid/cpp/src/tests/Makefile.am
@@ -28,7 +28,8 @@ TESTS+=unit_test
 check_PROGRAMS+=unit_test
 unit_test_LDADD=-lboost_unit_test_framework -lboost_regex \
 	$(lib_client) $(lib_broker) 
-unit_test_SOURCES= unit_test.cpp unit_test.h BrokerFixture.h \
+unit_test_SOURCES= unit_test.cpp unit_test.h
+	BrokerFixture.h SocketProxy.h \
 	exception_test.cpp \
 	RefCounted.cpp RefCountedMap.cpp \
 	SessionState.cpp Blob.cpp logging.cpp \
diff --git a/qpid/cpp/src/tests/SocketProxy.h b/qpid/cpp/src/tests/SocketProxy.h
new file mode 100644
index 0000000000..03d9b6ad35
--- /dev/null
+++ b/qpid/cpp/src/tests/SocketProxy.h
@@ -0,0 +1,80 @@
+#ifndef SOCKETPROXY_H
+#define SOCKETPROXY_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/Socket.h"
+#include "qpid/sys/Runnable.h"
+#include "qpid/sys/Thread.h"
+
+/**
+ * A simple socket proxy that forwards to another socket. Used between
+ * client & broker to simulate network failures.
+ */
+struct SocketProxy : public qpid::sys::Runnable
+{
+    int port;             // Port bound to server socket.
+    qpid::sys::Socket client, server; // Client & server sockets.
+
+    SocketProxy(const std::string& host, int port) { init(host,port); }
+    SocketProxy(int port) { init("localhost",port); }
+
+    ~SocketProxy() { client.close(); server.close(); thread.join(); }
+    
+  private:
+
+    void init(const std::string& host, int port) {
+        client.connect(host,port);
+        port = server.listen();
+        thread=qpid::sys::Thread(this);
+    }
+
+    void run() {
+        do {
+            ssize_t recv = server.recv(buffer, sizeof(buffer));
+            if (recv <= 0) return;
+            ssize_t sent=client.send(buffer, recv);
+            if (sent < 0) return;
+            assert(sent == recv); // Assumes we can send as we receive.
+        } while (true);
+    }
+
+    qpid::sys::Thread thread;
+    char buffer[64*1024];
+};
+
+/** A local client connection via a socket proxy. */
+struct ProxyConnection : public qpid::client::Connection {
+    SocketProxy proxy;
+    qpid::client::Session_0_10 session;
+    
+    ProxyConnection(const std::string& host, int port) : proxy(port) {
+        open(host, proxy.port);
+        session=newSession();
+    }
+
+    ProxyConnection(int port) : proxy(port) {
+        open("localhost", proxy.port);
+        session=newSession();
+    }
+};
+
+#endif
diff --git a/qpid/cpp/src/tests/exception_test.cpp b/qpid/cpp/src/tests/exception_test.cpp
index e774c48070..a8ee3bd37c 100644
--- a/qpid/cpp/src/tests/exception_test.cpp
+++ b/qpid/cpp/src/tests/exception_test.cpp
@@ -21,6 +21,7 @@
 
 #include "unit_test.h"
 #include "BrokerFixture.h"
+#include "SocketProxy.h"
 #include "qpid/client/SubscriptionManager.h"
 #include "qpid/sys/Runnable.h"
 #include "qpid/sys/Thread.h"
@@ -71,31 +72,33 @@ struct Catcher : public Runnable {
 };
 
 BOOST_FIXTURE_TEST_CASE(DisconnectedGet, BrokerFixture) {
-    Catcher<ClosedException> get(bind(&Session_0_10::get, session));
-    disconnect(connection);
+    ProxyConnection c(broker->getPort());
+    Catcher<ClosedException> get(bind(&Session_0_10::get, c.session));
+    c.proxy.client.close();           // Close the client side.
     BOOST_CHECK(get.join());
 }
 
 BOOST_FIXTURE_TEST_CASE(DisconnectedPop, BrokerFixture) {
-    session.queueDeclare(arg::queue="q");
+    ProxyConnection c(broker->getPort());
+    c.session.queueDeclare(arg::queue="q");
     subs.subscribe(lq, "q");
     Catcher<ClosedException> pop(bind(&LocalQueue::pop, boost::ref(lq)));
-    disconnect(connection);
+    c.proxy.client.close();
     BOOST_CHECK(pop.join());
 }
 
-// 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;
-//     session.queueDeclare(arg::queue="q");
-//     subs.subscribe(l, "q");
-//     Thread t(subs);
-//     disconnect(connection);
-//     t.join();
-//     BOOST_CHECK_THROW(session.close(), InternalErrorException);    
-// }
+BOOST_FIXTURE_TEST_CASE(DisconnectedListen, BrokerFixture) {
+    struct NullListener : public MessageListener {
+        void received(Message&) { BOOST_FAIL("Unexpected message"); }
+    } l;
+    ProxyConnection c;
+    c.session.queueDeclare(arg::queue="q");
+    subs.subscribe(l, "q");
+    Thread t(subs);
+    c.proxy.client.close();
+    t.join();
+    BOOST_CHECK_THROW(c.session.close(), InternalErrorException);    
+}
 
 BOOST_FIXTURE_TEST_CASE(NoSuchQueueTest, BrokerFixture) {
     BOOST_CHECK_THROW(subs.subscribe(lq, "no such queue").sync(), NotFoundException);