QPID-2328 - Applied patch from Ian Main

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

4 files changed, 26 insertions, 4 deletions

diff --git a/qpid/cpp/bindings/qmf/python/qmf.py b/qpid/cpp/bindings/qmf/python/qmf.py
index eec975c50f..e4ab581dfd 100644
--- a/qpid/cpp/bindings/qmf/python/qmf.py
+++ b/qpid/cpp/bindings/qmf/python/qmf.py
@@ -242,8 +242,7 @@ class Connection(Thread):
 
 
     def kick(self):
-        self._sockEngine.send(".")
-        # self._sockEngine.flush()  Not available with python?
+        self.impl.notify()
 
 
     def add_conn_handler(self, handler):
diff --git a/qpid/cpp/bindings/qmf/ruby/qmf.rb b/qpid/cpp/bindings/qmf/ruby/qmf.rb
index cc2aadc337..ce824b3605 100644
--- a/qpid/cpp/bindings/qmf/ruby/qmf.rb
+++ b/qpid/cpp/bindings/qmf/ruby/qmf.rb
@@ -216,8 +216,7 @@ module Qmf
     end
 
     def kick
-      @sockEngine.write(".")
-      @sockEngine.flush
+      @impl.notify
     end
 
     def add_conn_handler(handler)
diff --git a/qpid/cpp/include/qmf/engine/ResilientConnection.h b/qpid/cpp/include/qmf/engine/ResilientConnection.h
index 359c8ea6ff..c03d08cb96 100644
--- a/qpid/cpp/include/qmf/engine/ResilientConnection.h
+++ b/qpid/cpp/include/qmf/engine/ResilientConnection.h
@@ -155,6 +155,14 @@ namespace engine {
          */
         void setNotifyFd(int fd);
 
+        /**
+         * Send a byte into the notify file descriptor.
+         *
+         * This can be used to wake up the event processing portion of the engine from either the
+         * wrapped implementation or the engine itself.
+         */
+        void notify();
+
     private:
         ResilientConnectionImpl* impl;
     };
diff --git a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp
index 9c19e4d460..ab65b8d768 100644
--- a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp
+++ b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp
@@ -96,6 +96,7 @@ namespace engine {
         void bind(SessionHandle handle, char* exchange, char* queue, char* key);
         void unbind(SessionHandle handle, char* exchange, char* queue, char* key);
         void setNotifyFd(int fd);
+        void notify();
 
         void run();
         void failure();
@@ -329,6 +330,16 @@ void ResilientConnectionImpl::unbind(SessionHandle handle,
     sess->session.exchangeUnbind(client::arg::exchange=exchange, client::arg::queue=queue, client::arg::bindingKey=key);
 }
 
+void ResilientConnectionImpl::notify()
+{
+    if (notifyFd != -1)
+    {
+        int unused_ret;    //Suppress warnings about ignoring return value.
+        unused_ret = ::write(notifyFd, ".", 1);
+    }
+}
+
+
 void ResilientConnectionImpl::setNotifyFd(int fd)
 {
     notifyFd = fd;
@@ -496,3 +507,8 @@ void ResilientConnection::setNotifyFd(int fd)
     impl->setNotifyFd(fd);
 }
 
+void ResilientConnection::notify()
+{
+    impl->notify();
+}
+