Serialise close into the data callbacks:

Rejig Rdma::ConnectionManager to have a stop function with a callback and
use this to ensure that the Rdma::Connector used by qpid::sys::RdmaConnector
is correctly deleted only after it has been actually stopped


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

2 files changed, 22 insertions, 4 deletions

diff --git a/cpp/src/qpid/sys/rdma/RdmaIO.cpp b/cpp/src/qpid/sys/rdma/RdmaIO.cpp
index fe7062d3ea..23660a0b9f 100644
--- a/cpp/src/qpid/sys/rdma/RdmaIO.cpp
+++ b/cpp/src/qpid/sys/rdma/RdmaIO.cpp
@@ -518,6 +518,7 @@ namespace Rdma {
         ErrorCallback errc,
         DisconnectedCallback dc
     ) :
+        state(IDLE),
         ci(Connection::make()),
         handle(*ci, boost::bind(&ConnectionManager::event, this, _1), 0, 0),
         errorCallback(errc),
@@ -537,11 +538,23 @@ namespace Rdma {
         handle.startWatch(poller);
     }
 
-    void ConnectionManager::stop() {
+    void ConnectionManager::doStoppedCallback() {
+        // Ensure we can't get any more callbacks (except for the stopped callback)
         handle.stopWatch();
+
+        NotifyCallback nc;
+        nc.swap(notifyCallback);
+        nc(*this);
+    }
+
+    void ConnectionManager::stop(NotifyCallback nc) {
+        state = STOPPED;
+        notifyCallback = nc;
+        handle.call(boost::bind(&ConnectionManager::doStoppedCallback, this));
     }
 
     void ConnectionManager::event(DispatchHandle&) {
+        if (state.get() == STOPPED) return;
         connectionEvent(ci);
     }
 
diff --git a/cpp/src/qpid/sys/rdma/RdmaIO.h b/cpp/src/qpid/sys/rdma/RdmaIO.h
index 55174ea8a1..00eba28716 100644
--- a/cpp/src/qpid/sys/rdma/RdmaIO.h
+++ b/cpp/src/qpid/sys/rdma/RdmaIO.h
@@ -26,7 +26,6 @@
 #include "qpid/sys/AtomicValue.h"
 #include "qpid/sys/Dispatcher.h"
 #include "qpid/sys/DispatchHandle.h"
-#include "qpid/sys/Mutex.h"
 #include "qpid/sys/SocketAddress.h"
 
 #include <netinet/in.h>
@@ -163,14 +162,19 @@ namespace Rdma {
     typedef boost::function1<void, Rdma::Connection::intrusive_ptr> DisconnectedCallback;
 
     class ConnectionManager {
+        typedef boost::function1<void, ConnectionManager&> NotifyCallback;
+ 
+        enum State {IDLE, STOPPED};
+        qpid::sys::AtomicValue<State> state;
         Connection::intrusive_ptr ci;
         qpid::sys::DispatchHandleRef handle;
+        NotifyCallback notifyCallback;
 
     protected:
         ErrorCallback errorCallback;
         DisconnectedCallback disconnectedCallback;
 
-   public:
+    public:
         ConnectionManager(
             ErrorCallback errc,
             DisconnectedCallback dc
@@ -179,10 +183,11 @@ namespace Rdma {
         virtual ~ConnectionManager();
 
         void start(qpid::sys::Poller::shared_ptr poller, const qpid::sys::SocketAddress& addr);
-        void stop();
+        void stop(NotifyCallback);
 
     private:
         void event(qpid::sys::DispatchHandle& handle);
+        void doStoppedCallback();
 
         virtual void startConnection(Connection::intrusive_ptr ci, const qpid::sys::SocketAddress& addr) = 0;
         virtual void connectionEvent(Connection::intrusive_ptr ci) = 0;