Fix RDMA for upstream changes which now require notification on shutdown

differently from before

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

1 files changed, 15 insertions, 5 deletions

diff --git a/cpp/src/qpid/sys/RdmaIOPlugin.cpp b/cpp/src/qpid/sys/RdmaIOPlugin.cpp
index 58317838bc..e3498fad47 100644
--- a/cpp/src/qpid/sys/RdmaIOPlugin.cpp
+++ b/cpp/src/qpid/sys/RdmaIOPlugin.cpp
@@ -74,6 +74,7 @@ class RdmaIOHandler : public OutputControl {
     void full(Rdma::AsynchIO& aio);
     void idle(Rdma::AsynchIO& aio);
     void error(Rdma::AsynchIO& aio);
+    void drained(Rdma::AsynchIO& aio);
 };
 
 RdmaIOHandler::RdmaIOHandler(Rdma::Connection::intrusive_ptr& c, qpid::sys::ConnectionCodec::Factory* f) :
@@ -89,12 +90,18 @@ void RdmaIOHandler::init(Rdma::AsynchIO* a) {
     aio = a;
 }
 
+namespace {
+    void deleteAsynchIO(Rdma::AsynchIO& aio) {
+        delete &aio;
+    }
+}
+
 RdmaIOHandler::~RdmaIOHandler() {
     if (codec)
         codec->closed();
     delete codec;
 
-    aio->deferDelete();
+    aio->stop(deleteAsynchIO);
 }
 
 void RdmaIOHandler::write(const framing::ProtocolInitiation& data)
@@ -108,7 +115,7 @@ void RdmaIOHandler::write(const framing::ProtocolInitiation& data)
 }
 
 void RdmaIOHandler::close() {
-    aio->queueWriteClose();
+    aio->drainWriteQueue(boost::bind(&RdmaIOHandler::drained, this, _1));
 }
 
 // TODO: Dummy implementation, need to fill this in for heartbeat timeout to work
@@ -133,7 +140,7 @@ void RdmaIOHandler::idle(Rdma::AsynchIO&) {
         aio->queueWrite(buff);
     }
     if (codec->isClosed())
-        aio->queueWriteClose();
+        aio->drainWriteQueue(boost::bind(&RdmaIOHandler::drained, this, _1));
 }
 
 void RdmaIOHandler::initProtocolOut() {
@@ -149,6 +156,9 @@ void RdmaIOHandler::error(Rdma::AsynchIO&) {
     close();
 }
 
+void RdmaIOHandler::drained(Rdma::AsynchIO&) {
+}
+
 void RdmaIOHandler::full(Rdma::AsynchIO&) {
     QPID_LOG(debug, "Rdma: buffer full [" << identifier << "]");
 }
@@ -176,7 +186,7 @@ void RdmaIOHandler::readbuff(Rdma::AsynchIO&, Rdma::Buffer* buff) {
     }catch(const std::exception& e){
         QPID_LOG(error, e.what());
         readError = true;
-        aio->queueWriteClose();
+        aio->drainWriteQueue(boost::bind(&RdmaIOHandler::drained, this, _1));
     }
 }
 
@@ -195,7 +205,7 @@ void RdmaIOHandler::initProtocolIn(Rdma::Buffer* buff) {
             // send valid version header & close connection.
             write(framing::ProtocolInitiation(framing::highestProtocolVersion));
             readError = true;
-            aio->queueWriteClose();
+            aio->drainWriteQueue(boost::bind(&RdmaIOHandler::drained, this, _1));
         }
     }
 }