QPID-1050: Patch from Ted Ross:

1) Durability for federation links (broker-to-broker connections)
2) Improved handling of federation links:
    a) Links can be created even if the remote broker is not reachable
    b) If links are lost, re-establishment will occur using an exponential back-off algorithm
3) Durability of exchanges is now viewable through management
4) ManagementAgent API has been moved to an interface class to reduce coupling between the broker and manageable plug-ins.
5) General configuration storage capability has been added to the store/recover interface. This is used for federation links.
6) Management object-ids for durable objects are now themselves durable. 

(Note: some refactoring needed around ProtocolAccess needed to try and reduce dependencies)



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

1 files changed, 102 insertions, 0 deletions

diff --git a/cpp/src/qpid/broker/LinkRegistry.cpp b/cpp/src/qpid/broker/LinkRegistry.cpp
new file mode 100644
index 0000000000..6e20a3f7ce
--- /dev/null
+++ b/cpp/src/qpid/broker/LinkRegistry.cpp
@@ -0,0 +1,102 @@
+/*
+ *
+ * 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 "LinkRegistry.h"
+#include <iostream>
+
+using namespace qpid::broker;
+using namespace qpid::sys;
+using std::pair;
+using std::stringstream;
+using boost::intrusive_ptr;
+
+#define LINK_MAINT_INTERVAL 5
+
+LinkRegistry::LinkRegistry (Broker* _broker) : broker(_broker), parent(0), store(0)
+{
+    timer.add (intrusive_ptr<TimerTask> (new Periodic(*this)));
+}
+
+LinkRegistry::Periodic::Periodic (LinkRegistry& _links) :
+    TimerTask (Duration (LINK_MAINT_INTERVAL * TIME_SEC)), links(_links) {}
+
+void LinkRegistry::Periodic::fire ()
+{
+    links.periodicMaintenance ();
+    links.timer.add (intrusive_ptr<TimerTask> (new Periodic(links)));
+}
+
+void LinkRegistry::periodicMaintenance ()
+{
+    Mutex::ScopedLock locker(lock);
+    linksToDestroy.clear();
+    for (LinkMap::iterator i = links.begin(); i != links.end(); i++)
+        i->second->maintenanceVisit();
+}
+
+pair<Link::shared_ptr, bool> LinkRegistry::declare(std::string& host,
+                                                   uint16_t     port,
+                                                   bool         useSsl,
+                                                   bool         durable)
+{
+    Mutex::ScopedLock   locker(lock);
+    stringstream        keystream;
+    keystream << host << ":" << port;
+    string key = string(keystream.str());
+
+    LinkMap::iterator i = links.find(key);
+    if (i == links.end())
+    {
+        Link::shared_ptr link;
+
+        link = Link::shared_ptr (new Link (this, host, port, useSsl, durable, broker, parent));
+        links[key] = link;
+        return std::pair<Link::shared_ptr, bool>(link, true);
+    }
+    return std::pair<Link::shared_ptr, bool>(i->second, false);
+}
+
+void LinkRegistry::destroy(const string& host, const uint16_t port)
+{
+    Mutex::ScopedLock   locker(lock);
+    stringstream        keystream;
+    keystream << host << ":" << port;
+    string key = string(keystream.str());
+
+    LinkMap::iterator i = links.find(key);
+    if (i != links.end())
+    {
+        if (i->second->isDurable() && store)
+            store->destroy(*(i->second));
+        linksToDestroy[key] = i->second;
+        links.erase(i);
+    }
+}
+
+void LinkRegistry::setStore (MessageStore* _store)
+{
+    assert (store == 0 && _store != 0);
+    store = _store;
+}
+
+MessageStore* LinkRegistry::getStore() const {
+    return store;
+}
+