NO-JIRA: eliminate SocketProxy from tests.

Was causing problems with changes to introduce IPv6.
In most cases this was being used unnecessarilly due to cut-and-paste of tests.
In the 2 cases it was useful, replaced by simply shutting down the broker
to simulate network failure.

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

9 files changed, 28 insertions, 253 deletions

diff --git a/cpp/src/tests/BrokerFixture.h b/cpp/src/tests/BrokerFixture.h
index 672d954572..92c6d22b57 100644
--- a/cpp/src/tests/BrokerFixture.h
+++ b/cpp/src/tests/BrokerFixture.h
@@ -22,8 +22,6 @@
  *
  */
 
-#include "SocketProxy.h"
-
 #include "qpid/broker/Broker.h"
 #include "qpid/client/Connection.h"
 #include "qpid/client/ConnectionImpl.h"
@@ -71,16 +69,15 @@ struct  BrokerFixture : private boost::noncopyable {
         brokerThread = qpid::sys::Thread(*broker);
     };
 
-    void shutdownBroker()
-    {
-        broker->shutdown();
-        broker = BrokerPtr();
+    void shutdownBroker() {
+        if (broker) {
+            broker->shutdown();
+            brokerThread.join();
+            broker = BrokerPtr();
+        }
     }
 
-    ~BrokerFixture() {
-        if (broker) broker->shutdown();
-        brokerThread.join();
-    }
+    ~BrokerFixture() {  shutdownBroker(); }
 
     /** Open a connection to the broker. */
     void open(qpid::client::Connection& c) {
@@ -97,20 +94,6 @@ struct LocalConnection : public qpid::client::Connection {
     ~LocalConnection() { close(); }
 };
 
-/** A local client connection via a socket proxy. */
-struct ProxyConnection : public qpid::client::Connection {
-    SocketProxy proxy;
-    ProxyConnection(int brokerPort) : proxy(brokerPort) {
-        open("localhost", proxy.getPort());
-    }
-    ProxyConnection(const qpid::client::ConnectionSettings& s) : proxy(s.port) {
-        qpid::client::ConnectionSettings proxySettings(s);
-        proxySettings.port = proxy.getPort();
-        open(proxySettings);
-    }
-    ~ProxyConnection() { close(); }
-};
-
 /** Convenience class to create and open a connection and session
  * and some related useful objects.
  */
@@ -147,7 +130,6 @@ struct  SessionFixtureT : BrokerFixture, ClientT<ConnectionType,SessionType> {
 };
 
 typedef SessionFixtureT<LocalConnection> SessionFixture;
-typedef SessionFixtureT<ProxyConnection> ProxySessionFixture;
 
 }} // namespace qpid::tests
 
diff --git a/cpp/src/tests/ClientSessionTest.cpp b/cpp/src/tests/ClientSessionTest.cpp
index 3c0cff7350..30441cd03c 100644
--- a/cpp/src/tests/ClientSessionTest.cpp
+++ b/cpp/src/tests/ClientSessionTest.cpp
@@ -102,9 +102,9 @@ struct SimpleListener : public MessageListener
     }
 };
 
-struct ClientSessionFixture : public ProxySessionFixture
+struct ClientSessionFixture : public SessionFixture
 {
-    ClientSessionFixture(Broker::Options opts = Broker::Options()) : ProxySessionFixture(opts) {
+    ClientSessionFixture(Broker::Options opts = Broker::Options()) : SessionFixture(opts) {
         session.queueDeclare(arg::queue="my-queue");
     }
 };
@@ -150,16 +150,6 @@ QPID_AUTO_TEST_CASE(testDispatcherThread)
         BOOST_CHECK_EQUAL(boost::lexical_cast<string>(i), listener.messages[i].getData());
 }
 
-// FIXME aconway 2009-06-17: test for unimplemented feature, enable when implemented.
-void testSuspend0Timeout() {
-    ClientSessionFixture fix;
-    fix.session.suspend();  // session has 0 timeout.
-    try {
-        fix.connection.resume(fix.session);
-        BOOST_FAIL("Expected InvalidArgumentException.");
-    } catch(const InternalErrorException&) {}
-}
-
 QPID_AUTO_TEST_CASE(testUseSuspendedError)
 {
     ClientSessionFixture fix;
@@ -171,18 +161,6 @@ QPID_AUTO_TEST_CASE(testUseSuspendedError)
     } catch(const NotAttachedException&) {}
 }
 
-// FIXME aconway 2009-06-17: test for unimplemented feature, enable when implemented.
-void testSuspendResume() {
-    ClientSessionFixture fix;
-    fix.session.timeout(60);
-    fix.session.suspend();
-    // Make sure we are still subscribed after resume.
-    fix.connection.resume(fix.session);
-    fix.session.messageTransfer(arg::content=Message("my-message", "my-queue"));
-    BOOST_CHECK_EQUAL("my-message", fix.subs.get("my-queue", TIME_SEC).getData());
-}
-
-
 QPID_AUTO_TEST_CASE(testSendToSelf) {
     ClientSessionFixture fix;
     SimpleListener mylistener;
diff --git a/cpp/src/tests/MessageReplayTracker.cpp b/cpp/src/tests/MessageReplayTracker.cpp
index 3d79ee53c2..e35f673683 100644
--- a/cpp/src/tests/MessageReplayTracker.cpp
+++ b/cpp/src/tests/MessageReplayTracker.cpp
@@ -51,7 +51,7 @@ class ReplayBufferChecker
 
 QPID_AUTO_TEST_CASE(testReplay)
 {
-    ProxySessionFixture fix;
+    SessionFixture fix;
     fix.session.queueDeclare(arg::queue="my-queue", arg::exclusive=true, arg::autoDelete=true);
 
     MessageReplayTracker tracker(10);
@@ -77,7 +77,7 @@ QPID_AUTO_TEST_CASE(testReplay)
 
 QPID_AUTO_TEST_CASE(testCheckCompletion)
 {
-    ProxySessionFixture fix;
+    SessionFixture fix;
     fix.session.queueDeclare(arg::queue="my-queue", arg::exclusive=true, arg::autoDelete=true);
 
     MessageReplayTracker tracker(10);
diff --git a/cpp/src/tests/QueueEvents.cpp b/cpp/src/tests/QueueEvents.cpp
index bd18fa45fb..cea8bbf0db 100644
--- a/cpp/src/tests/QueueEvents.cpp
+++ b/cpp/src/tests/QueueEvents.cpp
@@ -147,7 +147,7 @@ struct EventRecorder
 
 QPID_AUTO_TEST_CASE(testSystemLevelEventProcessing)
 {
-    ProxySessionFixture fixture;
+    SessionFixture fixture;
     //register dummy event listener to broker
     EventRecorder listener;
     fixture.broker->getQueueEvents().registerListener("recorder", boost::bind(&EventRecorder::handle, &listener, _1));
@@ -194,7 +194,7 @@ QPID_AUTO_TEST_CASE(testSystemLevelEventProcessing)
 
 QPID_AUTO_TEST_CASE(testSystemLevelEventProcessing_enqueuesOnly)
 {
-    ProxySessionFixture fixture;
+    SessionFixture fixture;
     //register dummy event listener to broker
     EventRecorder listener;
     fixture.broker->getQueueEvents().registerListener("recorder", boost::bind(&EventRecorder::handle, &listener, _1));
diff --git a/cpp/src/tests/QueuePolicyTest.cpp b/cpp/src/tests/QueuePolicyTest.cpp
index 5455105078..f735e09449 100644
--- a/cpp/src/tests/QueuePolicyTest.cpp
+++ b/cpp/src/tests/QueuePolicyTest.cpp
@@ -152,7 +152,7 @@ QPID_AUTO_TEST_CASE(testRingPolicyCount)
     std::auto_ptr<QueuePolicy> policy = QueuePolicy::createQueuePolicy("test", 5, 0, QueuePolicy::RING);
     policy->update(args);
 
-    ProxySessionFixture f;
+    SessionFixture f;
     std::string q("my-ring-queue");
     f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args);
     for (int i = 0; i < 10; i++) {
@@ -187,7 +187,7 @@ QPID_AUTO_TEST_CASE(testRingPolicySize)
     std::auto_ptr<QueuePolicy> policy = QueuePolicy::createQueuePolicy("test", 0, 500, QueuePolicy::RING);
     policy->update(args);
 
-    ProxySessionFixture f;
+    SessionFixture f;
     std::string q("my-ring-queue");
     f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args);
 
@@ -259,7 +259,7 @@ QPID_AUTO_TEST_CASE(testStrictRingPolicy)
     std::auto_ptr<QueuePolicy> policy = QueuePolicy::createQueuePolicy("test", 5, 0, QueuePolicy::RING_STRICT);
     policy->update(args);
 
-    ProxySessionFixture f;
+    SessionFixture f;
     std::string q("my-ring-queue");
     f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args);
     LocalQueue incoming;
@@ -285,7 +285,7 @@ QPID_AUTO_TEST_CASE(testPolicyWithDtx)
     std::auto_ptr<QueuePolicy> policy = QueuePolicy::createQueuePolicy("test", 5, 0, QueuePolicy::REJECT);
     policy->update(args);
 
-    ProxySessionFixture f;
+    SessionFixture f;
     std::string q("my-policy-queue");
     f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args);
     LocalQueue incoming;
@@ -345,7 +345,7 @@ QPID_AUTO_TEST_CASE(testFlowToDiskWithNoStore)
     // Disable flow control, or else we'll never hit the max limit
     args.setInt(QueueFlowLimit::flowStopCountKey, 0);
 
-    ProxySessionFixture f;
+    SessionFixture f;
     std::string q("my-queue");
     f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args);
     LocalQueue incoming;
@@ -371,7 +371,7 @@ QPID_AUTO_TEST_CASE(testPolicyFailureOnCommit)
     std::auto_ptr<QueuePolicy> policy = QueuePolicy::createQueuePolicy("test", 5, 0, QueuePolicy::REJECT);
     policy->update(args);
 
-    ProxySessionFixture f;
+    SessionFixture f;
     std::string q("q");
     f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args);
     f.session.txSelect();
@@ -388,7 +388,7 @@ QPID_AUTO_TEST_CASE(testCapacityConversion)
     args.setString("qpid.max_count", "5");
     args.setString("qpid.flow_stop_count", "0");
 
-    ProxySessionFixture f;
+    SessionFixture f;
     std::string q("q");
     f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args);
     for (int i = 0; i < 5; i++) {
diff --git a/cpp/src/tests/ReplicationTest.cpp b/cpp/src/tests/ReplicationTest.cpp
index 7310a3fe20..1219a6b59e 100644
--- a/cpp/src/tests/ReplicationTest.cpp
+++ b/cpp/src/tests/ReplicationTest.cpp
@@ -74,7 +74,7 @@ QPID_AUTO_TEST_CASE(testReplicationExchange)
 {
     qpid::broker::Broker::Options brokerOpts(getBrokerOpts(list_of<string>("qpidd")
                                                            ("--replication-exchange-name=qpid.replication")));
-    ProxySessionFixture f(brokerOpts);
+    SessionFixture f(brokerOpts);
 
 
     std::string dataQ("queue-1");
diff --git a/cpp/src/tests/SocketProxy.h b/cpp/src/tests/SocketProxy.h
deleted file mode 100644
index d195f11aa9..0000000000
--- a/cpp/src/tests/SocketProxy.h
+++ /dev/null
@@ -1,183 +0,0 @@
-#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/IOHandle.h"
-#ifdef _WIN32
-#  include "qpid/sys/windows/IoHandlePrivate.h"
-   typedef SOCKET FdType;
-#else
-#  include "qpid/sys/posix/PrivatePosix.h"
-   typedef int FdType;
-#endif
-#include "qpid/sys/Socket.h"
-#include "qpid/sys/Runnable.h"
-#include "qpid/sys/Thread.h"
-#include "qpid/sys/Mutex.h"
-#include "qpid/log/Statement.h"
-
-#include <boost/lexical_cast.hpp>
-
-namespace qpid {
-namespace tests {
-
-/**
- * A simple socket proxy that forwards to another socket.
- * Used between client & local broker to simulate network failures.
- */
-class SocketProxy : private qpid::sys::Runnable
-{
-    // Need a Socket we can get the fd from
-    class LowSocket : public qpid::sys::Socket {
-    public:
-#ifdef _WIN32
-        FdType getFd() { return toSocketHandle(*this); }
-#else
-        FdType getFd() { return toFd(impl); }
-#endif
-    };
-
-  public:
-    /** Connect to connectPort on host, start a forwarding thread.
-     * Listen for connection on getPort().
-     */
-    SocketProxy(int connectPort, const std::string host="localhost")
-      : closed(false), joined(true),
-        port(listener.listen()), dropClient(), dropServer()
-    {
-        client.connect(host, boost::lexical_cast<std::string>(connectPort));
-        joined = false;
-        thread = qpid::sys::Thread(static_cast<qpid::sys::Runnable*>(this));
-    }
-
-      ~SocketProxy() { close(); if (!joined) thread.join(); }
-
-    /** Simulate a network disconnect. */
-    void close() {
-        {
-            qpid::sys::Mutex::ScopedLock l(lock);
-            if (closed) { return; }
-            closed=true;
-        }
-        if (thread && thread != qpid::sys::Thread::current()) {
-            thread.join();
-            joined = true;
-        }
-        client.close();
-    }
-
-    /** Simulate lost packets, drop data from client */
-    void dropClientData(bool drop=true) { dropClient=drop; }
-
-    /** Simulate lost packets, drop data from server */
-    void dropServerData(bool drop=true) { dropServer=drop; }
-
-    bool isClosed() const {
-        qpid::sys::Mutex::ScopedLock l(lock);
-        return closed;
-    }
-
-    uint16_t getPort() const { return port; }
-
-  private:
-    static void throwErrno(const std::string& msg) {
-        throw qpid::Exception(msg+":"+qpid::sys::strError(errno));
-    }
-    static void throwIf(bool condition, const std::string& msg) {
-        if (condition) throw qpid::Exception(msg);
-    }
-
-    void run() {
-        std::auto_ptr<LowSocket> server;
-        try {
-            fd_set socks;
-            FdType maxFd = listener.getFd();
-            struct timeval tmo;
-            for (;;) {
-                FD_ZERO(&socks);
-                FD_SET(maxFd, &socks);
-                tmo.tv_sec = 0;
-                tmo.tv_usec = 500 * 1000;
-                if (select(maxFd+1, &socks, 0, 0, &tmo) == 0) {
-                    qpid::sys::Mutex::ScopedLock l(lock);
-                    throwIf(closed, "SocketProxy: Closed by close()");
-                    continue;
-                }
-                throwIf(!FD_ISSET(maxFd, &socks), "SocketProxy: Accept failed");
-                break;   // Accept ready... go to next step
-            }
-            server.reset(reinterpret_cast<LowSocket *>(listener.accept()));
-            maxFd = server->getFd();
-            if (client.getFd() > maxFd)
-                maxFd = client.getFd();
-            char buffer[1024];
-            for (;;) {
-                FD_ZERO(&socks);
-                tmo.tv_sec = 0;
-                tmo.tv_usec = 500 * 1000;
-                FD_SET(client.getFd(), &socks);
-                FD_SET(server->getFd(), &socks);
-                if (select(maxFd+1, &socks, 0, 0, &tmo) == 0) {
-                    qpid::sys::Mutex::ScopedLock l(lock);
-                    throwIf(closed, "SocketProxy: Closed by close()");
-                    continue;
-                }
-                // Something is set; relay data as needed until something closes
-                if (FD_ISSET(server->getFd(), &socks)) {
-                    int n = server->read(buffer, sizeof(buffer));
-                    throwIf(n <= 0, "SocketProxy: server disconnected");
-                    if (!dropServer) client.write(buffer, n);
-                }
-                if (FD_ISSET(client.getFd(), &socks)) {
-                    int n = client.read(buffer, sizeof(buffer));
-                    throwIf(n <= 0, "SocketProxy: client disconnected");
-                    if (!dropServer) server->write(buffer, n);
-                }
-                if (!FD_ISSET(client.getFd(), &socks) &&
-                    !FD_ISSET(server->getFd(), &socks))
-                    throwIf(true, "SocketProxy: No handle ready");
-            }
-        }
-        catch (const std::exception& e) {
-            QPID_LOG(debug, "SocketProxy::run exception: " << e.what());
-        }
-        try {
-            if (server.get()) server->close();
-            close();
-        }
-        catch (const std::exception& e) {
-            QPID_LOG(debug, "SocketProxy::run exception in client/server close()" << e.what());
-        }
-    }
-
-    mutable qpid::sys::Mutex lock;
-    mutable bool closed;
-    bool joined;
-    LowSocket client, listener;
-    uint16_t port;
-    qpid::sys::Thread thread;
-    bool dropClient, dropServer;
-};
-
-}} // namespace qpid::tests
-
-#endif
diff --git a/cpp/src/tests/XmlClientSessionTest.cpp b/cpp/src/tests/XmlClientSessionTest.cpp
index b3b7f12b53..b94c35ece0 100644
--- a/cpp/src/tests/XmlClientSessionTest.cpp
+++ b/cpp/src/tests/XmlClientSessionTest.cpp
@@ -90,7 +90,7 @@ struct SimpleListener : public MessageListener
     }
 };
 
-struct ClientSessionFixture : public ProxySessionFixture
+struct ClientSessionFixture : public SessionFixture
 {
     void declareSubscribe(const string& q="odd_blue",
                           const string& dest="xml")
diff --git a/cpp/src/tests/exception_test.cpp b/cpp/src/tests/exception_test.cpp
index 3536ffddbe..3e844b4e58 100644
--- a/cpp/src/tests/exception_test.cpp
+++ b/cpp/src/tests/exception_test.cpp
@@ -92,32 +92,30 @@ QPID_AUTO_TEST_CASE(TestSessionBusy) {
 }
 
 QPID_AUTO_TEST_CASE(DisconnectedPop) {
-    ProxySessionFixture fix;
-    ProxyConnection c(fix.broker->getPort(Broker::TCP_TRANSPORT));
+    SessionFixture fix;
     fix.session.queueDeclare(arg::queue="q");
     fix.subs.subscribe(fix.lq, "q");
     Catcher<TransportFailure> pop(bind(&LocalQueue::pop, &fix.lq, sys::TIME_SEC));
-    fix.connection.proxy.close();
+    fix.shutdownBroker();
     BOOST_CHECK(pop.join());
 }
 
 QPID_AUTO_TEST_CASE(DisconnectedListen) {
-    ProxySessionFixture fix;
+    SessionFixture fix;
     struct NullListener : public MessageListener {
         void received(Message&) { BOOST_FAIL("Unexpected message"); }
     } l;
-    ProxyConnection c(fix.broker->getPort(Broker::TCP_TRANSPORT));
     fix.session.queueDeclare(arg::queue="q");
     fix.subs.subscribe(l, "q");
 
     Catcher<TransportFailure> runner(bind(&SubscriptionManager::run, boost::ref(fix.subs)));
-    fix.connection.proxy.close();
-    runner.join();    
+    fix.shutdownBroker();
+    runner.join();
     BOOST_CHECK_THROW(fix.session.queueDeclare(arg::queue="x"), TransportFailure);
 }
 
 QPID_AUTO_TEST_CASE(NoSuchQueueTest) {
-    ProxySessionFixture fix;
+    SessionFixture fix;
     ScopedSuppressLogging sl; // Suppress messages for expected errors.
     BOOST_CHECK_THROW(fix.subs.subscribe(fix.lq, "no such queue"), NotFoundException);
 }