Separate FailoverListener from client::Connection.

client::ConnectionImpl used to contain a FailoverListener to subscribe
for updates on the amq.failover exchange. This caused some lifecycle
issues including memory leaks.

Now FailoverListener is a public API class that the user must create
associated with a session to get known-broker updates.

Removed the weak_ptr logic in client::SessionImpl which was only
required because of FailoverListener.

Made SessionImpl::close() idempotent. Gets rid of spurious warning
messages in some tests.


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

4 files changed, 17 insertions, 11 deletions

diff --git a/cpp/src/tests/BrokerFixture.h b/cpp/src/tests/BrokerFixture.h
index 861400f539..f56a925b81 100644
--- a/cpp/src/tests/BrokerFixture.h
+++ b/cpp/src/tests/BrokerFixture.h
@@ -126,8 +126,8 @@ struct ClientT {
     ClientT(const qpid::client::ConnectionSettings& settings, const std::string& name_=std::string())
         : connection(settings), session(connection.newSession(name_)), subs(session), name(name_) {}
 
-    ~ClientT() { connection.close(); }
-    void close() { session.close(); connection.close(); }
+    ~ClientT() { close(); }
+    void close() { if (connection.isOpen()) { session.close(); connection.close(); } }
 };
 
 typedef ClientT<> Client;
diff --git a/cpp/src/tests/ClusterFixture.cpp b/cpp/src/tests/ClusterFixture.cpp
index 7c357c3cd1..fd90ed170e 100644
--- a/cpp/src/tests/ClusterFixture.cpp
+++ b/cpp/src/tests/ClusterFixture.cpp
@@ -141,13 +141,14 @@ void ClusterFixture::killWithSilencer(size_t n, client::Connection& c, int sig)
  * Get the known broker ports from a Connection.
  *@param n if specified wait for the cluster size to be n, up to a timeout.
  */
-std::set<int> knownBrokerPorts(qpid::client::Connection& source, int n) {
-    std::vector<qpid::Url> urls = source.getKnownBrokers();
+std::set<int> knownBrokerPorts(qpid::client::Connection& c, int n) {
+    FailoverListener fl(c);
+    std::vector<qpid::Url> urls = fl.getKnownBrokers();
     if (n >= 0 && unsigned(n) != urls.size()) {
         // Retry up to 10 secs in .1 second intervals.
         for (size_t retry=100; urls.size() != unsigned(n) && retry != 0; --retry) {
             qpid::sys::usleep(1000*100); // 0.1 secs
-            urls = source.getKnownBrokers();
+            urls = fl.getKnownBrokers();
         }
     }
     std::set<int> s;
diff --git a/cpp/src/tests/cluster_test.cpp b/cpp/src/tests/cluster_test.cpp
index 247aef1b2a..de8ec49ea6 100644
--- a/cpp/src/tests/cluster_test.cpp
+++ b/cpp/src/tests/cluster_test.cpp
@@ -358,13 +358,15 @@ QPID_AUTO_TEST_CASE(testTxTransaction) {
     rollbackSession.txRollback();
     rollbackSession.messageRelease(rollbackMessage.getId());
 
-
     // Verify queue status: just the comitted messages and dequeues should remain.
     BOOST_CHECK_EQUAL(c1.session.queueQuery("q").getMessageCount(), 4u);
     BOOST_CHECK_EQUAL(c1.subs.get("q", TIMEOUT).getData(), "B");
     BOOST_CHECK_EQUAL(c1.subs.get("q", TIMEOUT).getData(), "a");
     BOOST_CHECK_EQUAL(c1.subs.get("q", TIMEOUT).getData(), "b");
     BOOST_CHECK_EQUAL(c1.subs.get("q", TIMEOUT).getData(), "c");
+
+    commitSession.close();
+    rollbackSession.close();
 }
 
 QPID_AUTO_TEST_CASE(testUnacked) {
@@ -859,9 +861,12 @@ QPID_AUTO_TEST_CASE(testHeartbeatCancelledOnFailover)
     Receiver receiver(fmgr, "my-queue", "my-data");
     qpid::sys::Thread runner(receiver);
     receiver.waitForReady();
-    cluster.kill(1);
-    //sleep for 2 secs to allow the heartbeat task to fire on the now dead connection:
-    ::usleep(2*1000*1000);
+    {
+        ScopedSuppressLogging allQuiet; // suppress connection closed messages
+        cluster.kill(1);
+        //sleep for 2 secs to allow the heartbeat task to fire on the now dead connection:
+        ::usleep(2*1000*1000);
+    }
     fmgr.execute(sender);
     runner.join();
     BOOST_CHECK(!receiver.failed);
diff --git a/cpp/src/tests/exception_test.cpp b/cpp/src/tests/exception_test.cpp
index 0e9a948f00..4dac8ee965 100644
--- a/cpp/src/tests/exception_test.cpp
+++ b/cpp/src/tests/exception_test.cpp
@@ -112,8 +112,8 @@ QPID_AUTO_TEST_CASE(DisconnectedListen) {
 
     Catcher<TransportFailure> runner(bind(&SubscriptionManager::run, boost::ref(fix.subs)));
     fix.connection.proxy.close();
-    runner.join();
-    BOOST_CHECK_THROW(fix.session.close(), TransportFailure);
+    runner.join();    
+    BOOST_CHECK_THROW(fix.session.queueDeclare(arg::queue="x"), TransportFailure);
 }
 
 QPID_AUTO_TEST_CASE(NoSuchQueueTest) {