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

10 files changed, 56 insertions, 167 deletions

diff --git a/cpp/src/qpid/client/Connection.cpp b/cpp/src/qpid/client/Connection.cpp
index ae1cfa5815..32a01b2c40 100644
--- a/cpp/src/qpid/client/Connection.cpp
+++ b/cpp/src/qpid/client/Connection.cpp
@@ -152,8 +152,8 @@ void Connection::close() {
         impl->close();
 }
 
-std::vector<Url> Connection::getKnownBrokers() {
-    return impl ? impl->getKnownBrokers() : std::vector<Url>();
+std::vector<Url> Connection::getInitialBrokers() {
+    return impl ? impl->getInitialBrokers() : std::vector<Url>();
 }
 
 }} // namespace qpid::client
diff --git a/cpp/src/qpid/client/ConnectionImpl.cpp b/cpp/src/qpid/client/ConnectionImpl.cpp
index 6a46cb6249..c48a580fe8 100644
--- a/cpp/src/qpid/client/ConnectionImpl.cpp
+++ b/cpp/src/qpid/client/ConnectionImpl.cpp
@@ -22,7 +22,6 @@
 #include "qpid/client/Connector.h"
 #include "qpid/client/ConnectionSettings.h"
 #include "qpid/client/SessionImpl.h"
-#include "qpid/client/FailoverListener.h"
 
 #include "qpid/log/Statement.h"
 #include "qpid/Url.h"
@@ -88,7 +87,6 @@ ConnectionImpl::~ConnectionImpl() {
     // Important to close the connector first, to ensure the
     // connector thread does not call on us while the destructor
     // is running.
-    failover.reset();
     if (connector) connector->close();
 }
 
@@ -175,7 +173,6 @@ void ConnectionImpl::open()
     } else {
         QPID_LOG(debug, "No security layer in place");
     }
-    failover.reset(new FailoverListener(shared_from_this(), handler.knownBrokersUrls));
 }
 
 void ConnectionImpl::idleIn()
@@ -256,8 +253,8 @@ const ConnectionSettings& ConnectionImpl::getNegotiatedSettings()
     return handler;
 }
 
-std::vector<qpid::Url> ConnectionImpl::getKnownBrokers() {
-    return failover ? failover->getKnownBrokers() : handler.knownBrokersUrls;
+std::vector<qpid::Url> ConnectionImpl::getInitialBrokers() {
+    return handler.knownBrokersUrls;
 }
 
 boost::shared_ptr<SessionImpl>  ConnectionImpl::newSession(const std::string& name, uint32_t timeout, uint16_t channel) {
@@ -267,6 +264,4 @@ boost::shared_ptr<SessionImpl>  ConnectionImpl::newSession(const std::string& na
     return simpl;
 }
 
-void ConnectionImpl::stopFailoverListener() { failover->stop(); }
-
 }} // namespace qpid::client
diff --git a/cpp/src/qpid/client/ConnectionImpl.h b/cpp/src/qpid/client/ConnectionImpl.h
index 8c82b6a3fe..2b32e1ccf0 100644
--- a/cpp/src/qpid/client/ConnectionImpl.h
+++ b/cpp/src/qpid/client/ConnectionImpl.h
@@ -42,7 +42,6 @@ namespace client {
 class Connector;
 struct ConnectionSettings;
 class SessionImpl;
-class FailoverListener;
 
 class ConnectionImpl : public Bounds,
                        public framing::FrameHandler,
@@ -58,7 +57,6 @@ class ConnectionImpl : public Bounds,
     SessionMap sessions; 
     ConnectionHandler handler;
     boost::scoped_ptr<Connector> connector;
-    boost::scoped_ptr<FailoverListener> failover;
     framing::ProtocolVersion version;
     uint16_t nextChannel;
     sys::Mutex lock;
@@ -90,9 +88,8 @@ class ConnectionImpl : public Bounds,
     void erase(uint16_t channel);
     const ConnectionSettings& getNegotiatedSettings();
 
-    std::vector<Url> getKnownBrokers();
+    std::vector<Url> getInitialBrokers();
     void registerFailureCallback ( boost::function<void ()> fn ) { failureCallback = fn; }
-    void stopFailoverListener();
 
     framing::ProtocolVersion getVersion() { return version; }
 };
diff --git a/cpp/src/qpid/client/FailoverListener.cpp b/cpp/src/qpid/client/FailoverListener.cpp
index d84525ed89..3396f5598c 100644
--- a/cpp/src/qpid/client/FailoverListener.cpp
+++ b/cpp/src/qpid/client/FailoverListener.cpp
@@ -19,11 +19,7 @@
  *
  */
 #include "qpid/client/FailoverListener.h"
-#include "qpid/client/SessionBase_0_10Access.h"
-#include "qpid/client/SessionImpl.h"
-#include "qpid/client/ConnectionImpl.h"
-#include "qpid/client/SubscriptionImpl.h"
-#include "qpid/client/SubscriptionManager.h"
+#include "qpid/client/Session.h"
 #include "qpid/framing/Uuid.h"
 #include "qpid/log/Statement.h"
 #include "qpid/log/Helpers.h"
@@ -31,83 +27,46 @@
 namespace qpid {
 namespace client {
 
-static const std::string AMQ_FAILOVER("amq.failover");
+const std::string FailoverListener::AMQ_FAILOVER("amq.failover");
 
-static Session makeSession(boost::shared_ptr<SessionImpl> si) {
-    // Hold only a weak pointer to the ConnectionImpl so a
-    // FailoverListener in a ConnectionImpl won't createa a shared_ptr
-    // cycle.
-    // 
-    si->setWeakPtr(true);
-    Session s;
-    SessionBase_0_10Access(s).set(si);
-    return s;
-}
-
-FailoverListener::FailoverListener(const boost::shared_ptr<ConnectionImpl>& c, const std::vector<Url>& initUrls)
-  : knownBrokers(initUrls) 
- {
-    // Special versions used to mark cluster catch-up connections
-    // which do not need a FailoverListener
-    if (c->getVersion().getMajor() >= 0x80)  {
-        QPID_LOG(debug, "No failover listener for catch-up connection.");
-        return;
-    }
-
-    Session session = makeSession(c->newSession(AMQ_FAILOVER+framing::Uuid(true).str(), 0));
+FailoverListener::FailoverListener(Connection c) :
+    connection(c),
+    session(c.newSession(AMQ_FAILOVER+"."+framing::Uuid(true).str())),
+    subscriptions(session)
+{
+    knownBrokers = c.getInitialBrokers();
     if (session.exchangeQuery(arg::name=AMQ_FAILOVER).getNotFound()) {
         session.close();
         return;
     }
-    subscriptions.reset(new SubscriptionManager(session));
     std::string qname=session.getId().getName();
     session.queueDeclare(arg::queue=qname, arg::exclusive=true, arg::autoDelete=true);
     session.exchangeBind(arg::queue=qname, arg::exchange=AMQ_FAILOVER);
-    subscriptions->subscribe(*this, qname, SubscriptionSettings(FlowControl::unlimited(), ACCEPT_MODE_NONE));
+    subscriptions.subscribe(*this, qname, SubscriptionSettings(FlowControl::unlimited(),
+                                                                ACCEPT_MODE_NONE));
     thread = sys::Thread(*this);
 }
 
-void FailoverListener::run() 
-{
+void FailoverListener::run() {
     try {
-        subscriptions->run();
-    } catch (const TransportFailure&) {
-    } catch (const std::exception& e) {
-        QPID_LOG(error, QPID_MSG(e.what()));
-    }
+        subscriptions.run();
+    } catch(...) {}
 }
 
 FailoverListener::~FailoverListener() {
-    try { stop(); }
-    catch (const std::exception& /*e*/) {}
-}
-
-void FailoverListener::stop() {
-    if (subscriptions.get()) 
-        subscriptions->stop();
-
-    if (thread.id() == sys::Thread::current().id()) {
-        // FIXME aconway 2008-10-16: this can happen if ConnectionImpl
-        // dtor runs when my session drops its weak pointer lock.
-        // For now, leak subscriptions to prevent a core if we delete
-        // without joining.
-        subscriptions.release();
-    }
-    else if (thread.id()) {
+    try {
+        subscriptions.stop();
         thread.join();
-        thread=sys::Thread();
-        subscriptions.reset();  // Safe to delete after join.
-    }
+        if (connection.isOpen()) {
+            session.sync();
+            session.close();
+        }
+    } catch (...) {}
 }
 
 void FailoverListener::received(Message& msg) {
     sys::Mutex::ScopedLock l(lock);
-    knownBrokers.clear();
-    framing::Array urlArray;
-    msg.getHeaders().getArray("amq.failover", urlArray);
-    for (framing::Array::ValueVector::const_iterator i = urlArray.begin(); i != urlArray.end(); ++i ) 
-        knownBrokers.push_back(Url((*i)->get<std::string>()));
-    QPID_LOG(info, "Known-brokers update: " << log::formatList(knownBrokers));
+    knownBrokers = getKnownBrokers(msg);
 }
 
 std::vector<Url> FailoverListener::getKnownBrokers() const {
@@ -115,4 +74,16 @@ std::vector<Url> FailoverListener::getKnownBrokers() const {
     return knownBrokers;
 }
 
+std::vector<Url> FailoverListener::getKnownBrokers(const Message& msg) {
+    std::vector<Url> knownBrokers;
+    framing::Array urlArray;
+    msg.getHeaders().getArray("amq.failover", urlArray);
+    for (framing::Array::ValueVector::const_iterator i = urlArray.begin();
+         i != urlArray.end();
+         ++i ) 
+        knownBrokers.push_back(Url((*i)->get<std::string>()));
+    return knownBrokers;
+}
+
+
 }} // namespace qpid::client
diff --git a/cpp/src/qpid/client/FailoverListener.h b/cpp/src/qpid/client/FailoverListener.h
deleted file mode 100644
index 7afee736ac..0000000000
--- a/cpp/src/qpid/client/FailoverListener.h
+++ /dev/null
@@ -1,60 +0,0 @@
-#ifndef QPID_CLIENT_FAILOVERLISTENER_H
-#define QPID_CLIENT_FAILOVERLISTENER_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/client/MessageListener.h"
-#include "qpid/Url.h"
-#include "qpid/sys/Mutex.h"
-#include "qpid/sys/Runnable.h"
-#include "qpid/sys/Thread.h"
-#include <vector>
-
-namespace qpid {
-namespace client {
-
-class SubscriptionManager;
-class ConnectionImpl;
-
-/**
- * @internal Listen for failover updates from the amq.failover exchange.
- */
-class FailoverListener : public MessageListener, private qpid::sys::Runnable 
-{
-  public:
-    FailoverListener(const boost::shared_ptr<ConnectionImpl>&, const std::vector<Url>& initUrls);
-    ~FailoverListener();
-    void stop();
-
-    std::vector<Url> getKnownBrokers() const;
-    void received(Message& msg);
-    void run();
-    
-  private:
-    mutable sys::Mutex lock;
-    std::auto_ptr<SubscriptionManager> subscriptions;
-    sys::Thread thread;
-    std::vector<Url> knownBrokers;
-};
-}} // namespace qpid::client
-
-#endif  /*!QPID_CLIENT_FAILOVERLISTENER_H*/
diff --git a/cpp/src/qpid/client/FailoverManager.cpp b/cpp/src/qpid/client/FailoverManager.cpp
index 967c3613c0..81f71eb7df 100644
--- a/cpp/src/qpid/client/FailoverManager.cpp
+++ b/cpp/src/qpid/client/FailoverManager.cpp
@@ -77,7 +77,9 @@ Connection& FailoverManager::connect(std::vector<Url> brokers)
         } else {
             state = CONNECTING;
             Connection c;
-            attempt(c, settings, brokers.empty() ? connection.getKnownBrokers() : brokers);
+            if (brokers.empty() && failoverListener.get())
+                brokers = failoverListener->getKnownBrokers();
+            attempt(c, settings, brokers);
             if (c.isOpen()) state = IDLE;
             else state = CANT_CONNECT;
             connection = c;
@@ -118,6 +120,7 @@ void FailoverManager::attempt(Connection& c, ConnectionSettings s)
     try {
         QPID_LOG(info, "Attempting to connect to " << s.host << " on " << s.port << "..."); 
         c.open(s);
+        failoverListener.reset(new FailoverListener(c));
         QPID_LOG(info, "Connected to " << s.host << " on " << s.port); 
     } catch (const Exception& e) {
         QPID_LOG(info, "Could not connect to " << s.host << " on " << s.port << ": " << e.what()); 
diff --git a/cpp/src/qpid/client/SessionImpl.cpp b/cpp/src/qpid/client/SessionImpl.cpp
index b3181bdb3c..7c807558f0 100644
--- a/cpp/src/qpid/client/SessionImpl.cpp
+++ b/cpp/src/qpid/client/SessionImpl.cpp
@@ -57,9 +57,7 @@ SessionImpl::SessionImpl(const std::string& name, boost::shared_ptr<ConnectionIm
       detachedLifetime(0),
       maxFrameSize(conn->getNegotiatedSettings().maxFrameSize),
       id(conn->getNegotiatedSettings().username, name.empty() ? Uuid(true).str() : name),
-      connectionShared(conn),
-      connectionWeak(conn),
-      weakPtr(false),
+      connection(conn),
       ioHandler(*this),
       proxy(ioHandler),
       nextIn(0),
@@ -68,7 +66,7 @@ SessionImpl::SessionImpl(const std::string& name, boost::shared_ptr<ConnectionIm
       doClearDeliveryPropertiesExchange(true),
       autoDetach(true)
 {
-    channel.next = connectionShared.get();
+    channel.next = connection.get();
 }
 
 SessionImpl::~SessionImpl() {
@@ -87,8 +85,7 @@ SessionImpl::~SessionImpl() {
         }
         delete sendMsgCredit;
     }
-    boost::shared_ptr<ConnectionImpl> c =  connectionWeak.lock();
-    if (c) c->erase(channel);
+    connection->erase(channel);
 }
 
 
@@ -122,6 +119,7 @@ void SessionImpl::open(uint32_t timeout) // user thread
 void SessionImpl::close() //user thread
 {
     Lock l(state);
+    if (state == DETACHED || state == DETACHING) return;
     if (detachedLifetime) setTimeout(0);
     detach();
     waitFor(DETACHED);
@@ -129,8 +127,6 @@ void SessionImpl::close() //user thread
 
 void SessionImpl::resume(boost::shared_ptr<ConnectionImpl>) // user thread
 {
-    // weakPtr sessions should not be resumed.
-    if (weakPtr) return;
     throw NotImplementedException("Resume not yet implemented by client!");
 }
 
@@ -509,11 +505,8 @@ void SessionImpl::proxyOut(AMQFrame& frame) // network thread
 
 void SessionImpl::sendFrame(AMQFrame& frame, bool canBlock)
 {
-    boost::shared_ptr<ConnectionImpl> c =  connectionWeak.lock();
-    if (c) {
-        channel.handle(frame);
-        c->expand(frame.encodedSize(), canBlock);
-    }
+    channel.handle(frame);
+    connection->expand(frame.encodedSize(), canBlock);
 }
 
 void SessionImpl::deliver(AMQFrame& frame) // network thread
@@ -809,17 +802,9 @@ uint32_t SessionImpl::getTimeout() const {
     return detachedLifetime;
 }
 
-void SessionImpl::setWeakPtr(bool weak) {
-    weakPtr = weak;
-    if (weakPtr)
-        connectionShared.reset();   // Only keep weak pointer
-    else
-        connectionShared = connectionWeak.lock();
-}
-
 boost::shared_ptr<ConnectionImpl> SessionImpl::getConnection()
 {
-    return connectionWeak.lock();
+    return connection;
 }
 
 void SessionImpl::disableAutoDetach() { autoDetach = false; }
diff --git a/cpp/src/qpid/client/SessionImpl.h b/cpp/src/qpid/client/SessionImpl.h
index cbd0742045..2f35032c4e 100644
--- a/cpp/src/qpid/client/SessionImpl.h
+++ b/cpp/src/qpid/client/SessionImpl.h
@@ -120,11 +120,6 @@ public:
     /** Get timeout in seconds. */
     uint32_t getTimeout() const;
 
-    /** Make this session use a weak_ptr to the ConnectionImpl.
-     * Used for sessions created by the ConnectionImpl itself.
-     */
-    void setWeakPtr(bool weak=true);
-
     /** 
      * get the Connection associated with this connection
      */
@@ -224,9 +219,7 @@ private:
     const uint64_t maxFrameSize;
     const SessionId id;
 
-    boost::shared_ptr<ConnectionImpl> connectionShared;
-    boost::weak_ptr<ConnectionImpl> connectionWeak;
-    bool weakPtr;
+    boost::shared_ptr<ConnectionImpl> connection;
 
     framing::FrameHandler::MemFunRef<SessionImpl, &SessionImpl::proxyOut> ioHandler;
     framing::ChannelHandler channel;
diff --git a/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp b/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
index 3a735b5698..1698f96caf 100644
--- a/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
+++ b/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
@@ -136,7 +136,9 @@ void ConnectionImpl::connect(const AbsTime& started)
 
 bool ConnectionImpl::tryConnect()
 {
-    if (tryConnect(url) || tryConnect(connection.getKnownBrokers())) {
+    if (tryConnect(url) ||
+        (failoverListener.get() && tryConnect(failoverListener->getKnownBrokers())))
+    {
         return resetSessions();
     } else {
         return false;
@@ -148,6 +150,7 @@ bool ConnectionImpl::tryConnect(const Url& u)
     try {
         QPID_LOG(info, "Trying to connect to " << url << "...");                
         connection.open(u, settings);
+        failoverListener.reset(new FailoverListener(connection));
         return true;
     } catch (const Exception& e) {
         //TODO: need to fix timeout on open so that it throws TransportFailure
diff --git a/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h b/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h
index 565f2ec7ec..f4bc09594d 100644
--- a/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h
+++ b/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h
@@ -25,6 +25,7 @@
 #include "qpid/messaging/Variant.h"
 #include "qpid/Url.h"
 #include "qpid/client/Connection.h"
+#include "qpid/client/FailoverListener.h"
 #include "qpid/client/ConnectionSettings.h"
 #include "qpid/sys/Mutex.h"
 #include "qpid/sys/Semaphore.h"
@@ -50,6 +51,7 @@ class ConnectionImpl : public qpid::messaging::ConnectionImpl
     qpid::sys::Mutex lock;//used to protect data structures
     qpid::sys::Semaphore semaphore;//used to coordinate reconnection
     qpid::client::Connection connection;
+    std::auto_ptr<FailoverListener> failoverListener;
     qpid::Url url;
     qpid::client::ConnectionSettings settings;
     Sessions sessions;