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, 38 insertions, 16 deletions

diff --git a/cpp/src/qpid/sys/TCPIOPlugin.cpp b/cpp/src/qpid/sys/TCPIOPlugin.cpp
index 045bc56e90..5d2cadbe03 100644
--- a/cpp/src/qpid/sys/TCPIOPlugin.cpp
+++ b/cpp/src/qpid/sys/TCPIOPlugin.cpp
@@ -41,13 +41,15 @@ class AsynchIOProtocolFactory : public ProtocolFactory {
   public:
     AsynchIOProtocolFactory(int16_t port, int backlog);
     void accept(Poller::shared_ptr, ConnectionCodec::Factory*);
-    void connect(Poller::shared_ptr, const std::string& host, int16_t port, ConnectionCodec::Factory*);
+    void connect(Poller::shared_ptr, const std::string& host, int16_t port,
+                 ConnectionCodec::Factory*, ProtocolAccess*);
 
     uint16_t getPort() const;
     std::string getHost() const;
 
   private:
-    void established(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*, bool isClient);
+    void established(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*,
+                     bool isClient, ProtocolAccess*);
 };
 
 // Static instance to initialise plugin
@@ -72,17 +74,32 @@ AsynchIOProtocolFactory::AsynchIOProtocolFactory(int16_t port, int backlog) :
 {}
 
 void AsynchIOProtocolFactory::established(Poller::shared_ptr poller, const Socket& s,
-                                          ConnectionCodec::Factory* f, bool isClient) {
-    AsynchIOHandler* async = new AsynchIOHandler(s.getPeerAddress(), f);
+                                          ConnectionCodec::Factory* f, bool isClient,
+                                          ProtocolAccess* a) {
+    AsynchIOHandler* async = new AsynchIOHandler(s.getPeerAddress(), f, a);
+    AsynchIO* aio;
+
     if (isClient)
         async->setClient();
-    AsynchIO* aio = new AsynchIO(s,
-                                 boost::bind(&AsynchIOHandler::readbuff, async, _1, _2),
-                                 boost::bind(&AsynchIOHandler::eof, async, _1),
-                                 boost::bind(&AsynchIOHandler::disconnect, async, _1),
-                                 boost::bind(&AsynchIOHandler::closedSocket, async, _1, _2),
-                                 boost::bind(&AsynchIOHandler::nobuffs, async, _1),
-                                 boost::bind(&AsynchIOHandler::idle, async, _1));
+    if (a == 0)
+        aio = new AsynchIO(s,
+                           boost::bind(&AsynchIOHandler::readbuff, async, _1, _2),
+                           boost::bind(&AsynchIOHandler::eof, async, _1),
+                           boost::bind(&AsynchIOHandler::disconnect, async, _1),
+                           boost::bind(&AsynchIOHandler::closedSocket, async, _1, _2),
+                           boost::bind(&AsynchIOHandler::nobuffs, async, _1),
+                           boost::bind(&AsynchIOHandler::idle, async, _1));
+    else {
+        aio = new AsynchIO(s,
+                           boost::bind(&AsynchIOHandler::readbuff, async, _1, _2),
+                           boost::bind(&ProtocolAccess::closedEof, a, async),
+                           boost::bind(&AsynchIOHandler::disconnect, async, _1),
+                           boost::bind(&AsynchIOHandler::closedSocket, async, _1, _2),
+                           boost::bind(&AsynchIOHandler::nobuffs, async, _1),
+                           boost::bind(&AsynchIOHandler::idle, async, _1));
+        a->setAio(aio);
+    }
+
     async->init(aio, 4);
     aio->start(poller);
 }
@@ -95,26 +112,31 @@ std::string AsynchIOProtocolFactory::getHost() const {
     return listener.getSockname();
 }
 
-void AsynchIOProtocolFactory::accept(Poller::shared_ptr poller, ConnectionCodec::Factory* fact) {
+void AsynchIOProtocolFactory::accept(Poller::shared_ptr poller,
+                                     ConnectionCodec::Factory* fact) {
     acceptor.reset(
         new AsynchAcceptor(listener,
-                           boost::bind(&AsynchIOProtocolFactory::established, this, poller, _1, fact, false)));
+                           boost::bind(&AsynchIOProtocolFactory::established, this, poller, _1, fact, false,
+                                       (ProtocolAccess*) 0)));
     acceptor->start(poller);
 }
 
 void AsynchIOProtocolFactory::connect(
     Poller::shared_ptr poller,
     const std::string& host, int16_t port,
-    ConnectionCodec::Factory* f)
+    ConnectionCodec::Factory* fact,
+    ProtocolAccess* access)
 {
     // Note that the following logic does not cause a memory leak.
     // The allocated Socket is freed either by the AsynchConnector
     // upon connection failure or by the AsynchIO upon connection
     // shutdown.  The allocated AsynchConnector frees itself when it
     // is no longer needed.
+
     Socket* socket = new Socket();
-    new AsynchConnector(*socket, poller, host, port,
-                        boost::bind(&AsynchIOProtocolFactory::established, this, poller, _1, f, true));
+    new AsynchConnector (*socket, poller, host, port,
+                         boost::bind(&AsynchIOProtocolFactory::established, this, poller, _1, fact, true, access),
+                         boost::bind(&ProtocolAccess::closed, access, _1, _2));
 }
 
 }} // namespace qpid::sys