reworked fix of r827747 (QPID-2149)

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

2 files changed, 66 insertions, 29 deletions

diff --git a/qpid/python/qpid/compat.py b/qpid/python/qpid/compat.py
index 53ab757e89..e38f0dbe8c 100644
--- a/qpid/python/qpid/compat.py
+++ b/qpid/python/qpid/compat.py
@@ -42,3 +42,64 @@ if tuple(sys.version_info[0:2]) < (2, 4):
     return old_select(list(rlist), list(wlist), list(xlist), timeout)
 else:
   from select import select
+
+if sys.platform in ('win32', 'cygwin'):
+  import socket
+
+  class SockWaiter:
+
+    def __init__(self, read_sock, write_sock):
+      self.read_sock = read_sock
+      self.write_sock = write_sock
+
+    def wakeup(self):
+      self.write_sock.send("\0")
+
+    def fileno(self):
+      return self.read_sock.fileno()
+
+    def reading(self):
+      return True
+
+    def readable(self):
+      self.read_sock.recv(65536)
+
+  def __repr__(self):
+    return "SockWaiter(%r, %r)" % (self.read_sock, self.write_sock)
+
+  def selectable_waiter():
+    listener = socket.socket()
+    listener.bind(('', 0))
+    listener.listen(1)
+    _, port = listener.getsockname()
+    write_sock = socket.socket()
+    write_sock.connect(("127.0.0.1", port))
+    read_sock, _ = listener.accept()
+    listener.close()
+    return SockWaiter(read_sock, write_sock)
+else:
+  import os
+
+  class PipeWaiter:
+
+    def __init__(self, read_fd, write_fd):
+      self.read_fd = read_fd
+      self.write_fd = write_fd
+
+    def wakeup(self):
+      os.write(self.write_fd, "\0")
+
+    def fileno(self):
+      return self.read_fd
+
+    def reading(self):
+      return True
+
+    def readable(self):
+      os.read(self.read_fd, 65536)
+
+    def __repr__(self):
+      return "PipeWaiter(%r, %r)" % (self.read_fd, self.write_fd)
+
+  def selectable_waiter():
+    return PipeWaiter(*os.pipe())
diff --git a/qpid/python/qpid/selector.py b/qpid/python/qpid/selector.py
index a7993be0b1..ca5946c3f9 100644
--- a/qpid/python/qpid/selector.py
+++ b/qpid/python/qpid/selector.py
@@ -16,8 +16,8 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-import atexit, time, socket
-from compat import select, set
+import atexit, time
+from compat import select, set, selectable_waiter
 from threading import Thread, Lock
 
 class Acceptor:
@@ -39,23 +39,6 @@ class Acceptor:
     sock, addr = self.sock.accept()
     self.handler(sock)
 
-class Sink:
-
-  def __init__(self, sock):
-    self.sock = sock
-
-  def fileno(self):
-    return self.sock.fileno()
-
-  def reading(self):
-    return True
-
-  def readable(self):
-    self.sock.recv(65536)
-
-  def __repr__(self):
-    return "Sink(%r)" % self.sock.fileno()
-
 class Selector:
 
   lock = Lock()
@@ -78,20 +61,13 @@ class Selector:
     self.selectables = set()
     self.reading = set()
     self.writing = set()
-    listener = socket.socket()
-    listener.bind(('', 0))
-    listener.listen(1)
-    me_ip, me_port = listener.getsockname()
-    self.wakeup_sock = socket.socket()
-    self.wakeup_sock.connect(("127.0.0.1", me_port))
-    self.wait_sock, me = listener.accept()
-    listener.close()
-    self.reading.add(Sink(self.wait_sock))
+    self.waiter = selectable_waiter()
+    self.reading.add(self.waiter)
     self.stopped = False
     self.thread = None
 
   def wakeup(self):
-    self.wakeup_sock.send("\0")
+    self.waiter.wakeup()
 
   def register(self, selectable):
     self.selectables.add(selectable)