cleanup code

1 files changed, 15 insertions, 20 deletions

diff --git a/asyncio/subprocess_stream.py b/asyncio/subprocess_stream.py
index 4a8cbb1..8adee8c 100644
--- a/asyncio/subprocess_stream.py
+++ b/asyncio/subprocess_stream.py
@@ -2,6 +2,7 @@ __all__ = ['SubprocessStreamProtocol']
 
 from . import base_subprocess
 from . import events
+from . import protocols
 from . import streams
 from . import tasks
 
@@ -10,15 +11,14 @@ class WriteSubprocessPipeStreamProto(base_subprocess.WriteSubprocessPipeProto):
         base_subprocess.WriteSubprocessPipeProto.__init__(self, process_transport, fd)
         self._drain_waiter = None
         self._paused = False
-        self.writer = streams.StreamWriter(None, self, None, None)
+        self.writer = None
 
     def connection_made(self, transport):
         super().connection_made(transport)
-        self.writer._transport = transport
-        self.writer._loop = transport._loop
+        self.writer = streams.StreamWriter(transport, self, None, transport._loop)
 
     def connection_lost(self, exc):
-        # FIXME: call super().connection_lost(exc)
+        super().connection_lost(exc)
         # Also wake up the writing side.
         if self._paused:
             waiter = self._drain_waiter
@@ -44,7 +44,8 @@ class WriteSubprocessPipeStreamProto(base_subprocess.WriteSubprocessPipeProto):
                 waiter.set_result(None)
 
 
-class ReadSubprocessPipeStreamProto(base_subprocess.ReadSubprocessPipeProto):
+class ReadSubprocessPipeStreamProto(base_subprocess.WriteSubprocessPipeProto,
+                                    protocols.Protocol):
     def __init__(self, proc, fd, limit=streams._DEFAULT_LIMIT):
         super().__init__(proc, fd)
         self._stream_reader = streams.StreamReader(limit=limit)
@@ -54,7 +55,7 @@ class ReadSubprocessPipeStreamProto(base_subprocess.ReadSubprocessPipeProto):
         self._stream_reader.set_transport(transport)
 
     def connection_lost(self, exc):
-        # FIXME: call super().connection_lost(exc)
+        super().connection_lost(exc)
         if exc is None:
             self._stream_reader.feed_eof()
         else:
@@ -79,28 +80,22 @@ class SubprocessStreamProtocol(base_subprocess.SubprocessProtocol):
 
     def create_read_pipe_protocol(self, transport, fd):
         pipe = ReadSubprocessPipeStreamProto(transport, fd, self.limit)
-        if fd == 1:
-            self.stdout = pipe._stream_reader
-        elif fd == 2:
-            self.stderr = pipe._stream_reader
+        self._pipes[fd] = pipe
         return pipe
 
     def create_write_pipe_protocol(self, transport, fd):
         pipe = WriteSubprocessPipeStreamProto(transport, fd)
-        if fd == 0:
-            self.stdin = pipe.writer
+        self._pipes[fd] = pipe
         return pipe
 
     def connection_made(self, transport):
         self._transport = transport
-
-    def pipe_data_received(self, fd, data):
-        pipe = self._pipes[fd]
-        pipe.data_received(data)
-
-    def pipe_connection_lost(self, fd, exc):
-        pipe = self._pipes[fd]
-        pipe.connection_lost(exc)
+        if 0 in self._pipes:
+            self.stdin = self._pipes[0].writer
+        if 1 in self._pipes:
+            self.stdout = self._pipes[1]._stream_reader
+        if 2 in self._pipes:
+            self.stderr = self._pipes[1]._stream_reader
 
     @tasks.coroutine
     def wait(self):