[Python-ideas] Tulip / PEP 3156 - subprocess events

[Paul Moore]

On 22 January 2013 16:33, Guido van Rossum <guido at python.org> wrote:
> I am not actually very committed to a particular design for a subprocess
> transport. I'll happily leave it up to others to come up with a design and
> make it work on multiple platforms.

OK. I've written a pipe transport (event_loop.connect_read_pipe and
event_loop.connect_write_pipe) and modified the existing subprocess
test to use it. I've also added a small read/write test.

The code is in my bitbucket repository at https://bitbucket.org/pmoore/tulip.

I'm not very happy with the call-back based style of the read/write
test. I'm sure it would be much better written in an async style, but
I don't know how to do so. If anyone who understands the async style
better than I do can offer a translation, I'd be very grateful - I'd
like to see if the resulting code looks sufficiently clear. Here's the
relevant code. The biggest ugliness is the need for the two protocol
classes, which basically do nothing but (1) collect data received and
(2) ignore unwanted callbacks.

class DummyProto(protocols.Protocol):
    def __init__(self):
        pass
    def connection_made(self):
        pass
    def data_received(self, data):
        pass
    def eof_received(self):
        pass
    def connection_lost():
        pass

class MyCollector(protocols.Protocol):
    def __init__(self):
        self.data = []
    def connection_made(self):
        pass
    def data_received(self, data):
        self.data.append(data)
    def eof_received(self):
        pass
    def connection_lost():
        pass
    def get_data(self):
        return b''.join(self.data)

def testReadWrite(self):
    proc = Popen(['/bin/tr', 'a-z', 'A-Z'], stdin=PIPE, stdout=PIPE)
    rt, rp = yield from self.event_loop.connect_read_pipe(MyCollector,
proc.stdout)
    wt, wp = yield from self.event_loop.connect_read_pipe(DummyProto,
proc.stdin)
    def send_data():
        wt.write("hello, world")
        wt.write_eof()
    self.event_loop.call_soon(send_data)
    self.event_loop.run()
    self.assertEqual(rp.get_data(), b'HELLO, WORLD')

Paul