bpo-36889: Merge asyncio streams (GH-13251)

https://bugs.python.org/issue36889

Lib/asyncio/__init__.py

@@ -3,6 +3,7 @@
 # flake8: noqa
 
 import sys
+import warnings
 
 # This relies on each of the submodules having an __all__ variable.
 from .base_events import *
@@ -43,3 +44,40 @@ if sys.platform == 'win32':  # pragma: no cover
 else:
     from .unix_events import *  # pragma: no cover
     __all__ += unix_events.__all__
+
+
+__all__ += ('StreamReader', 'StreamWriter', 'StreamReaderProtocol')  # deprecated
+
+
+def __getattr__(name):
+    global StreamReader, StreamWriter, StreamReaderProtocol
+    if name == 'StreamReader':
+        warnings.warn("StreamReader is deprecated since Python 3.8 "
+                      "in favor of Stream, and scheduled for removal "
+                      "in Python 3.10",
+                      DeprecationWarning,
+                      stacklevel=2)
+        from .streams import StreamReader as sr
+        StreamReader = sr
+        return StreamReader
+    if name == 'StreamWriter':
+        warnings.warn("StreamWriter is deprecated since Python 3.8 "
+                      "in favor of Stream, and scheduled for removal "
+                      "in Python 3.10",
+                      DeprecationWarning,
+                      stacklevel=2)
+        from .streams import StreamWriter as sw
+        StreamWriter = sw
+        return StreamWriter
+    if name == 'StreamReaderProtocol':
+        warnings.warn("Using asyncio internal class StreamReaderProtocol "
+                      "is deprecated since Python 3.8 "
+                      " and scheduled for removal "
+                      "in Python 3.10",
+                      DeprecationWarning,
+                      stacklevel=2)
+        from .streams import StreamReaderProtocol as srp
+        StreamReaderProtocol = srp
+        return StreamReaderProtocol
+
+    raise AttributeError(f"module {__name__} has no attribute {name}")

Lib/asyncio/subprocess.py

@@ -27,6 +27,8 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
         self._process_exited = False
         self._pipe_fds = []
         self._stdin_closed = self._loop.create_future()
+        self._stdout_closed = self._loop.create_future()
+        self._stderr_closed = self._loop.create_future()
 
     def __repr__(self):
         info = [self.__class__.__name__]
@@ -40,10 +42,12 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
 
     def connection_made(self, transport):
         self._transport = transport
-
         stdout_transport = transport.get_pipe_transport(1)
         if stdout_transport is not None:
-            self.stdout = streams.StreamReader(limit=self._limit,
+            self.stdout = streams.Stream(mode=streams.StreamMode.READ,
+                                         transport=stdout_transport,
+                                         protocol=self,
+                                         limit=self._limit,
                                          loop=self._loop,
                                          _asyncio_internal=True)
             self.stdout.set_transport(stdout_transport)
@@ -51,7 +55,10 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
 
         stderr_transport = transport.get_pipe_transport(2)
         if stderr_transport is not None:
-            self.stderr = streams.StreamReader(limit=self._limit,
+            self.stderr = streams.Stream(mode=streams.StreamMode.READ,
+                                         transport=stderr_transport,
+                                         protocol=self,
+                                         limit=self._limit,
                                          loop=self._loop,
                                          _asyncio_internal=True)
             self.stderr.set_transport(stderr_transport)
@@ -59,9 +66,9 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
 
         stdin_transport = transport.get_pipe_transport(0)
         if stdin_transport is not None:
-            self.stdin = streams.StreamWriter(stdin_transport,
+            self.stdin = streams.Stream(mode=streams.StreamMode.WRITE,
+                                        transport=stdin_transport,
                                         protocol=self,
-                                              reader=None,
                                         loop=self._loop,
                                         _asyncio_internal=True)
 
@@ -114,6 +121,10 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
     def _get_close_waiter(self, stream):
         if stream is self.stdin:
             return self._stdin_closed
+        elif stream is self.stdout:
+            return self._stdout_closed
+        elif stream is self.stderr:
+            return self._stderr_closed
 
 
 class Process:

Lib/asyncio/windows_events.py

@@ -607,7 +607,7 @@ class IocpProactor:
 
             # ConnectPipe() failed with ERROR_PIPE_BUSY: retry later
             delay = min(delay * 2, CONNECT_PIPE_MAX_DELAY)
-            await tasks.sleep(delay, loop=self._loop)
+            await tasks.sleep(delay)
 
         return windows_utils.PipeHandle(handle)
 

Lib/test/test___all__.py

@@ -30,6 +30,10 @@ class AllTest(unittest.TestCase):
             raise NoAll(modname)
         names = {}
         with self.subTest(module=modname):
+            with support.check_warnings(
+                ("", DeprecationWarning),
+                ("", ResourceWarning),
+                quiet=True):
                 try:
                     exec("from %s import *" % modname, names)
                 except Exception as e:
@@ -40,6 +44,8 @@ class AllTest(unittest.TestCase):
                     del names["__builtins__"]
                 if '__annotations__' in names:
                     del names['__annotations__']
+                if "__warningregistry__" in names:
+                    del names["__warningregistry__"]
                 keys = set(names)
                 all_list = sys.modules[modname].__all__
                 all_set = set(all_list)

Lib/test/test_asyncio/test_base_events.py

@@ -1152,6 +1152,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
     @unittest.skipUnless(hasattr(socket, 'AF_INET6'), 'no IPv6 support')
     def test_create_server_ipv6(self):
         async def main():
+            with self.assertWarns(DeprecationWarning):
                 srv = await asyncio.start_server(
                     lambda: None, '::1', 0, loop=self.loop)
             try:

Lib/test/test_asyncio/test_buffered_proto.py

@@ -58,6 +58,7 @@ class BaseTestBufferedProtocol(func_tests.FunctionalTestCaseMixin):
             writer.close()
             await writer.wait_closed()
 
+        with self.assertWarns(DeprecationWarning):
             srv = self.loop.run_until_complete(
                 asyncio.start_server(
                     on_server_client, '127.0.0.1', 0))

Lib/test/test_asyncio/test_pep492.py

@@ -94,7 +94,9 @@ class StreamReaderTests(BaseTest):
     def test_readline(self):
         DATA = b'line1\nline2\nline3'
 
-        stream = asyncio.StreamReader(loop=self.loop, _asyncio_internal=True)
+        stream = asyncio.Stream(mode=asyncio.StreamMode.READ,
+                                loop=self.loop,
+                                _asyncio_internal=True)
         stream.feed_data(DATA)
         stream.feed_eof()
 

Lib/test/test_asyncio/test_server.py

@@ -46,6 +46,7 @@ class BaseStartServer(func_tests.FunctionalTestCaseMixin):
             async with srv:
                 await srv.serve_forever()
 
+        with self.assertWarns(DeprecationWarning):
             srv = self.loop.run_until_complete(asyncio.start_server(
                 serve, support.HOSTv4, 0, loop=self.loop, start_serving=False))
 
@@ -102,6 +103,7 @@ class SelectorStartServerTests(BaseStartServer, unittest.TestCase):
                 await srv.serve_forever()
 
         with test_utils.unix_socket_path() as addr:
+            with self.assertWarns(DeprecationWarning):
                 srv = self.loop.run_until_complete(asyncio.start_unix_server(
                     serve, addr, loop=self.loop, start_serving=False))
 

Lib/test/test_asyncio/test_sslproto.py

@@ -649,6 +649,7 @@ class BaseStartTLS(func_tests.FunctionalTestCaseMixin):
                 sock.close()
 
         async def client(addr):
+            with self.assertWarns(DeprecationWarning):
                 reader, writer = await asyncio.open_connection(
                     *addr,
                     ssl=client_sslctx,
@@ -688,6 +689,7 @@ class BaseStartTLS(func_tests.FunctionalTestCaseMixin):
                 sock.close()
 
         async def client(addr):
+            with self.assertWarns(DeprecationWarning):
                 reader, writer = await asyncio.open_connection(
                     *addr,
                     ssl=client_sslctx,
@@ -724,6 +726,7 @@ class BaseStartTLS(func_tests.FunctionalTestCaseMixin):
                 sock.close()
 
         async def client(addr):
+            with self.assertWarns(DeprecationWarning):
                 reader, writer = await asyncio.open_connection(
                     *addr,
                     ssl=client_sslctx,

Lib/test/test_asyncio/test_windows_events.py

@@ -17,6 +17,7 @@ import _winapi
 
 import asyncio
 from asyncio import windows_events
+from asyncio.streams import _StreamProtocol
 from test.test_asyncio import utils as test_utils
 from test.support.script_helper import spawn_python
 
@@ -100,16 +101,16 @@ class ProactorTests(test_utils.TestCase):
 
         clients = []
         for i in range(5):
-            stream_reader = asyncio.StreamReader(loop=self.loop,
+            stream = asyncio.Stream(mode=asyncio.StreamMode.READ,
-                                                 _asyncio_internal=True)
+                                    loop=self.loop, _asyncio_internal=True)
-            protocol = asyncio.StreamReaderProtocol(stream_reader,
+            protocol = _StreamProtocol(stream,
                                        loop=self.loop,
                                        _asyncio_internal=True)
             trans, proto = await self.loop.create_pipe_connection(
                 lambda: protocol, ADDRESS)
             self.assertIsInstance(trans, asyncio.Transport)
             self.assertEqual(protocol, proto)
-            clients.append((stream_reader, trans))
+            clients.append((stream, trans))
 
         for i, (r, w) in enumerate(clients):
             w.write('lower-{}\n'.format(i).encode())
@@ -118,6 +119,7 @@ class ProactorTests(test_utils.TestCase):
             response = await r.readline()
             self.assertEqual(response, 'LOWER-{}\n'.format(i).encode())
             w.close()
+            await r.close()
 
         server.close()
 

Misc/NEWS.d/next/Library/2019-05-14-12-25-44.bpo-36889.MChPqP.rst

@@ -0,0 +1,6 @@
+Introduce :class:`asyncio.Stream` class that merges :class:`asyncio.StreamReader` and :class:`asyncio.StreamWriter` functionality.
+:class:`asyncio.Stream` can work in readonly, writeonly and readwrite modes.
+Provide :func:`asyncio.connect`, :func:`asyncio.connect_unix`, :func:`asyncio.connect_read_pipe` and :func:`asyncio.connect_write_pipe` factories to open :class:`asyncio.Stream` connections. Provide :class:`asyncio.StreamServer` and :class:`UnixStreamServer` to serve servers with asyncio.Stream API.
+Modify :func:`asyncio.create_subprocess_shell` and :func:`asyncio.create_subprocess_exec` to use :class:`asyncio.Stream` instead of deprecated :class:`StreamReader` and :class:`StreamWriter`.
+Deprecate :class:`asyncio.StreamReader` and :class:`asyncio.StreamWriter`.
+Deprecate usage of private classes, e.g. :class:`asyncio.FlowControlMixing` and :class:`asyncio.StreamReaderProtocol` outside of asyncio package.