3 files changed, 23 insertions, 1 deletions
diff --git a/cpp/src/qpid/sys/RdmaIOPlugin.cpp b/cpp/src/qpid/sys/RdmaIOPlugin.cpp
index 8c7f410f00..b03f62337f 100644
--- a/cpp/src/qpid/sys/RdmaIOPlugin.cpp
+++ b/cpp/src/qpid/sys/RdmaIOPlugin.cpp
@@ -56,6 +56,7 @@ class RdmaIOHandler : public OutputControl {
     Rdma::Connection::intrusive_ptr connection;
 
     void write(const framing::ProtocolInitiation&);
+    void disconnectAction();
 
   public:
     RdmaIOHandler(Rdma::Connection::intrusive_ptr c, ConnectionCodec::Factory* f);
@@ -170,7 +171,7 @@ namespace {
     }
 }
 
-void RdmaIOHandler::disconnected() {
+void RdmaIOHandler::disconnectAction() {
     {
     Mutex::ScopedLock l(pollingLock);
     // If we're closed already then we'll get to drained() anyway
@@ -180,6 +181,10 @@ void RdmaIOHandler::disconnected() {
     aio->stop(boost::bind(&stopped, this));
 }
 
+void RdmaIOHandler::disconnected() {
+    aio->requestCallback(boost::bind(&RdmaIOHandler::disconnectAction, this));
+}
+
 void RdmaIOHandler::drained() {
     // We know we've drained the write queue now, but we don't have to do anything
     // because we can rely on the client to disconnect to trigger the connection
diff --git a/cpp/src/qpid/sys/rdma/RdmaIO.cpp b/cpp/src/qpid/sys/rdma/RdmaIO.cpp
index 068e8cf187..c89e0f2126 100644
--- a/cpp/src/qpid/sys/rdma/RdmaIO.cpp
+++ b/cpp/src/qpid/sys/rdma/RdmaIO.cpp
@@ -114,6 +114,20 @@ namespace Rdma {
         nc(*this);
     }
 
+    namespace {
+        void requestedCall(AsynchIO* aio, AsynchIO::RequestCallback callback) {
+            assert(callback);
+            callback(*aio);
+        }
+    }
+
+    void AsynchIO::requestCallback(RequestCallback callback) {
+        // TODO creating a function object every time isn't all that
+        // efficient - if this becomes heavily used do something better (what?)
+        assert(callback);
+        dataHandle.call(boost::bind(&requestedCall, this, callback));
+    }
+
     // Mark writing closed (so we don't accept any more writes or make any idle callbacks)
     void AsynchIO::drainWriteQueue(NotifyCallback nc) {
         State oldState;
diff --git a/cpp/src/qpid/sys/rdma/RdmaIO.h b/cpp/src/qpid/sys/rdma/RdmaIO.h
index 8d5dbc5f3e..72cbac154b 100644
--- a/cpp/src/qpid/sys/rdma/RdmaIO.h
+++ b/cpp/src/qpid/sys/rdma/RdmaIO.h
@@ -65,6 +65,8 @@ namespace Rdma {
         NotifyCallback notifyCallback;
 
     public:
+        typedef boost::function1<void, AsynchIO&> RequestCallback;
+
         // TODO: Instead of specifying a buffer size specify the amount of memory the AsynchIO class can use
         // for buffers both read and write (allocate half to each up front) and fail if we cannot allocate that much
         // locked memory
@@ -87,6 +89,7 @@ namespace Rdma {
         void notifyPendingWrite();
         void drainWriteQueue(NotifyCallback);
         void stop(NotifyCallback);
+        void requestCallback(RequestCallback);
         int incompletedWrites() const;
         Buffer* getBuffer();
         void returnBuffer(Buffer*);