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

3 files changed, 91 insertions, 4 deletions

diff --git a/cpp/include/qpid/client/Connection.h b/cpp/include/qpid/client/Connection.h
index 0f5999cdcc..bcf2962557 100644
--- a/cpp/include/qpid/client/Connection.h
+++ b/cpp/include/qpid/client/Connection.h
@@ -200,7 +200,11 @@ class Connection
 
     QPID_CLIENT_EXTERN bool isOpen() const;
 
-    QPID_CLIENT_EXTERN std::vector<Url> getKnownBrokers();
+    /** In a cluster, returns the initial set of known broker URLs
+     * at the time of connection.
+     */
+    QPID_CLIENT_EXTERN std::vector<Url> getInitialBrokers();
+
     QPID_CLIENT_EXTERN void registerFailureCallback ( boost::function<void ()> fn );
 
     /**
diff --git a/cpp/include/qpid/client/FailoverListener.h b/cpp/include/qpid/client/FailoverListener.h
new file mode 100644
index 0000000000..8414b80f2b
--- /dev/null
+++ b/cpp/include/qpid/client/FailoverListener.h
@@ -0,0 +1,81 @@
+#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/ClientImportExport.h"
+#include "qpid/client/MessageListener.h"
+#include "qpid/client/Connection.h"
+#include "qpid/client/Session.h"
+#include "qpid/client/SubscriptionManager.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 {
+
+
+/**
+ * Listen for updates from the amq.failover exchange.
+ *
+ * In a cluster, the amq.failover exchange provides updates whenever
+ * the cluster membership changes. This class subscribes to the
+ * failover exchange and providees the latest list of known brokers.
+ *
+ * You can also subscribe to amq.failover yourself and use
+ * FailoverListener::decode to extract a list of broker URLs from a
+ * failover exchange message.
+ */
+class FailoverListener : private MessageListener, private qpid::sys::Runnable 
+{
+  public:
+    /** The name of the standard failover exchange amq.failover */
+    static QPID_CLIENT_EXTERN const std::string AMQ_FAILOVER;
+
+    /** Extract the broker list from a failover exchange message */
+    static QPID_CLIENT_EXTERN std::vector<Url> getKnownBrokers(const Message& m);
+    
+    /** Subscribe to amq.failover exchange. */
+    QPID_CLIENT_EXTERN FailoverListener(Connection);
+
+    QPID_CLIENT_EXTERN ~FailoverListener();
+
+    /** Returns the latest list of known broker URLs. */
+    QPID_CLIENT_EXTERN std::vector<Url> getKnownBrokers() const;
+    
+  private:
+    void received(Message& msg);
+    void run();
+
+    mutable sys::Mutex lock;
+    Connection connection;
+    Session session;
+    SubscriptionManager subscriptions;
+    sys::Thread thread;
+    std::vector<Url> knownBrokers;
+};
+}} // namespace qpid::client
+
+#endif  /*!QPID_CLIENT_FAILOVERLISTENER_H*/
diff --git a/cpp/include/qpid/client/FailoverManager.h b/cpp/include/qpid/client/FailoverManager.h
index d50fcff8bb..0d30e2ed60 100644
--- a/cpp/include/qpid/client/FailoverManager.h
+++ b/cpp/include/qpid/client/FailoverManager.h
@@ -22,12 +22,13 @@
  *
  */
 
-#include "qpid/client/Connection.h"
-#include "qpid/client/ConnectionSettings.h"
 #include "qpid/Exception.h"
 #include "qpid/client/AsyncSession.h"
-#include "qpid/sys/Monitor.h"
 #include "qpid/client/ClientImportExport.h"
+#include "qpid/client/Connection.h"
+#include "qpid/client/ConnectionSettings.h"
+#include "qpid/client/FailoverListener.h"
+#include "qpid/sys/Monitor.h"
 #include <vector>
 
 namespace qpid {
@@ -123,6 +124,7 @@ class FailoverManager
 
     qpid::sys::Monitor lock;
     Connection connection;
+    std::auto_ptr<FailoverListener> failoverListener;
     ConnectionSettings settings;
     ReconnectionStrategy* strategy;
     State state;