Author: ritchiem
Date: Wed Oct 28 15:43:03 2009
New Revision: 830643
URL: http://svn.apache.org/viewvc?rev=830643&view=rev
Log:
reworked fix of r827747 (QPID-2149)
Modified:
qpid/branches/0.5.x-dev/qpid/python/qpid/compat.py
qpid/branches/0.5.x-dev/qpid/python/qpid/selector.py
Modified: qpid/branches/0.5.x-dev/qpid/python/qpid/compat.py
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/python/qpid/compat.py?rev=830643&r1=830642&r2=830643&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/python/qpid/compat.py (original)
+++ qpid/branches/0.5.x-dev/qpid/python/qpid/compat.py Wed Oct 28 15:43:03 2009
@@ -42,3 +42,64 @@
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())
Modified: qpid/branches/0.5.x-dev/qpid/python/qpid/selector.py
URL:
http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/python/qpid/selector.py?rev=830643&r1=830642&r2=830643&view=diff
==============================================================================
--- qpid/branches/0.5.x-dev/qpid/python/qpid/selector.py (original)
+++ qpid/branches/0.5.x-dev/qpid/python/qpid/selector.py Wed Oct 28 15:43:03
2009
@@ -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 @@
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 @@
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)
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscr...@qpid.apache.org
