make Popen work with gevent.wait()

- added rawlink() method to Popen
- added linkproxy helper class to gevent/hub.py
- modified examples/processes.py to use gevent.wait()

examples/processes.py

@@ -7,18 +7,14 @@ from gevent import subprocess
 p1 = subprocess.Popen(['uname'], stdout=subprocess.PIPE)
 p2 = subprocess.Popen(['ls'], stdout=subprocess.PIPE)
 
-job1 = gevent.spawn(p1.stdout.read)
-job2 = gevent.spawn(p2.stdout.read)
-
-# wait for them to complete. stop waiting after 2 seconds
-gevent.joinall([job1, job2], timeout=2)
+gevent.wait([p1, p2], timeout=2)
 
 # print the results (if available)
-if job1.ready():
-    print ('uname: %s bytes: %s' % (len(job1.value or ''), repr(job1.value)[:50]))
+if p1.poll() is not None:
+    print ('uname: %r' % p1.stdout.read())
 else:
     print ('uname: job is still running')
-if job2.ready():
-    print ('ls: %s bytes: %s' % (len(job2.value or ''), repr(job2.value)[:50]))
+if p2.poll() is not None:
+    print ('ls: %r' % p2.stdout.read())
 else:
     print ('ls: job is still running')

gevent/hub.py

@@ -635,6 +635,21 @@ def wait(objects=None, timeout=None, count=None):
     return result
 
 
+class linkproxy(object):
+    __slots__ = ['callback', 'obj']
+
+    def __init__(self, callback, obj):
+        self.callback = callback
+        self.obj = obj
+
+    def __call__(self, *args):
+        callback = self.callback
+        obj = self.obj
+        self.callback = None
+        self.obj = None
+        callback(obj)
+
+
 class _NONE(object):
     "A special thingy you must never pass to any of gevent API"
     __slots__ = []

gevent/subprocess.py

@@ -7,7 +7,7 @@ import gc
 import signal
 import traceback
 from gevent.event import AsyncResult
-from gevent.hub import get_hub
+from gevent.hub import get_hub, linkproxy
 from gevent.fileobject import FileObject
 from gevent.greenlet import Greenlet, joinall
 spawn = Greenlet.spawn
@@ -258,6 +258,10 @@ class Popen(object):
     def __repr__(self):
         return '<%s at 0x%x pid=%r returncode=%r>' % (self.__class__.__name__, id(self), self.pid, self.returncode)
 
+    def rawlink(self, callback):
+        self.result.rawlink(linkproxy(callback, self))
+    # XXX unlink
+
     def _on_child(self, watcher):
         watcher.stop()
         status = watcher.rstatus

greentest/test__subprocess.py

@@ -3,6 +3,7 @@ import sys
 import os
 import errno
 import greentest
+import gevent
 from gevent import subprocess
 import time
 
@@ -19,6 +20,11 @@ class Test(greentest.TestCase):
         popen = subprocess.Popen([sys.executable, '-c', 'import sys; sys.exit(10)'])
         self.assertEqual(popen.wait(), 10)
 
+    def test_wait(self):
+        popen = subprocess.Popen([sys.executable, '-c', 'import sys; sys.exit(11)'])
+        gevent.wait([popen])
+        self.assertEqual(popen.poll(), 11)
+
     def test_child_exception(self):
         try:
             subprocess.Popen(['*']).wait()