bpo-38242: Revert "bpo-36889: Merge asyncio streams (GH-13251)" (#16482)

See https://bugs.python.org/issue38242 for more details

Doc/library/asyncio-api-index.rst

@@ -132,47 +132,23 @@ High-level APIs to work with network IO.
     :widths: 50 50
     :class: full-width-table
 
-    * - ``await`` :func:`connect`
-      -  Establish a TCP connection to send and receive data.
-
     * - ``await`` :func:`open_connection`
-      -  Establish a TCP connection. (Deprecated in favor of :func:`connect`)
+      -  Establish a TCP connection.
-
-    * - ``await`` :func:`connect_unix`
-      -  Establish a Unix socket connection to send and receive data.
 
     * - ``await`` :func:`open_unix_connection`
-      -  Establish a Unix socket connection. (Deprecated in favor of :func:`connect_unix`)
+      -  Establish a Unix socket connection.
-
-    * - :class:`StreamServer`
-      - Start a TCP server.
 
     * - ``await`` :func:`start_server`
-      - Start a TCP server. (Deprecated in favor of :class:`StreamServer`)
+      - Start a TCP server.
-
-    * - :class:`UnixStreamServer`
-      - Start a Unix socket server.
 
     * - ``await`` :func:`start_unix_server`
-      - Start a Unix socket server. (Deprecated in favor of :class:`UnixStreamServer`)
+      - Start a Unix socket server.
-
-    * - :func:`connect_read_pipe`
-      - Establish a connection to :term:`file-like object <file object>` *pipe*
-        to receive data.
-
-    * - :func:`connect_write_pipe`
-      - Establish a connection to :term:`file-like object <file object>` *pipe*
-        to send data.
-
-    * - :class:`Stream`
-      - Stream is a single object combining APIs of :class:`StreamReader` and
-        :class:`StreamWriter`.
 
     * - :class:`StreamReader`
-      - High-level async/await object to receive network data. (Deprecated in favor of :class:`Stream`)
+      - High-level async/await object to receive network data.
 
     * - :class:`StreamWriter`
-      - High-level async/await object to send network data. (Deprecated in favor of :class:`Stream`)
+      - High-level async/await object to send network data.
 
 
 .. rubric:: Examples

Doc/library/asyncio-eventloop.rst

@@ -1637,7 +1637,8 @@ Wait until a file descriptor received some data using the
      :meth:`loop.create_connection` method.
 
    * Another similar :ref:`example <asyncio_example_create_connection-streams>`
-     using the high-level :func:`asyncio.connect` function and streams.
+     using the high-level :func:`asyncio.open_connection` function
+     and streams.
 
 
 .. _asyncio_example_unix_signals:

Doc/library/asyncio-protocol.rst

@@ -809,7 +809,7 @@ data, and waits until the connection is closed::
 .. seealso::
 
    The :ref:`TCP echo client using streams <asyncio-tcp-echo-client-streams>`
-   example uses the high-level :func:`asyncio.connect` function.
+   example uses the high-level :func:`asyncio.open_connection` function.
 
 
 .. _asyncio-udp-echo-server-protocol:
@@ -978,7 +978,7 @@ Wait until a socket receives data using the
 
    The :ref:`register an open socket to wait for data using streams
    <asyncio_example_create_connection-streams>` example uses high-level streams
-   created by the :func:`asyncio.connect` function in a coroutine.
+   created by the :func:`open_connection` function in a coroutine.
 
 .. _asyncio_example_subprocess_proto:
 

Doc/library/asyncio-stream.rst

@@ -18,12 +18,19 @@ streams::
     import asyncio
 
     async def tcp_echo_client(message):
-        async with asyncio.connect('127.0.0.1', 8888) as stream:
+        reader, writer = await asyncio.open_connection(
-            print(f'Send: {message!r}')
+            '127.0.0.1', 8888)
-            await stream.write(message.encode())
 
-            data = await stream.read(100)
+        print(f'Send: {message!r}')
-            print(f'Received: {data.decode()!r}')
+        writer.write(message.encode())
+        await writer.drain()
+
+        data = await reader.read(100)
+        print(f'Received: {data.decode()!r}')
+
+        print('Close the connection')
+        writer.close()
+        await writer.wait_closed()
 
     asyncio.run(tcp_echo_client('Hello World!'))
 
@@ -37,31 +44,6 @@ The following top-level asyncio functions can be used to create
 and work with streams:
 
 
-.. coroutinefunction:: connect(host=None, port=None, \*, \
-                               limit=2**16, ssl=None, family=0, \
-                               proto=0, flags=0, sock=None, local_addr=None, \
-                               server_hostname=None, ssl_handshake_timeout=None, \
-                               happy_eyeballs_delay=None, interleave=None)
-
-   Connect to TCP socket on *host* : *port* address and return a :class:`Stream`
-   object of mode :attr:`StreamMode.READWRITE`.
-
-   *limit* determines the buffer size limit used by the returned :class:`Stream`
-   instance. By default the *limit* is set to 64 KiB.
-
-   The rest of the arguments are passed directly to :meth:`loop.create_connection`.
-
-   The function can be used with ``await`` to get a connected stream::
-
-       stream = await asyncio.connect('127.0.0.1', 8888)
-
-   The function can also be used as an async context manager::
-
-       async with asyncio.connect('127.0.0.1', 8888) as stream:
-           ...
-
-   .. versionadded:: 3.8
-
 .. coroutinefunction:: open_connection(host=None, port=None, \*, \
                           loop=None, limit=None, ssl=None, family=0, \
                           proto=0, flags=0, sock=None, local_addr=None, \
@@ -87,12 +69,8 @@ and work with streams:
 
       The *ssl_handshake_timeout* parameter.
 
-   .. deprecated-removed:: 3.8 3.10
-
-      `open_connection()` is deprecated in favor of :func:`connect`.
-
 .. coroutinefunction:: start_server(client_connected_cb, host=None, \
-                          port=None, \*, loop=None, limit=2**16, \
+                          port=None, \*, loop=None, limit=None, \
                           family=socket.AF_UNSPEC, \
                           flags=socket.AI_PASSIVE, sock=None, \
                           backlog=100, ssl=None, reuse_address=None, \
@@ -124,60 +102,9 @@ and work with streams:
 
       The *ssl_handshake_timeout* and *start_serving* parameters.
 
-   .. deprecated-removed:: 3.8 3.10
-
-      `start_server()` is deprecated if favor of :class:`StreamServer`
-
-.. coroutinefunction:: connect_read_pipe(pipe, *, limit=2**16)
-
-   Takes a :term:`file-like object <file object>` *pipe* to return a
-   :class:`Stream` object of the mode :attr:`StreamMode.READ` that has
-   similar API of :class:`StreamReader`. It can also be used as an async context manager.
-
-   *limit* determines the buffer size limit used by the returned :class:`Stream`
-   instance. By default the limit is set to 64 KiB.
-
-   .. versionadded:: 3.8
-
-.. coroutinefunction:: connect_write_pipe(pipe, *, limit=2**16)
-
-   Takes a :term:`file-like object <file object>` *pipe* to return a
-   :class:`Stream` object of the mode :attr:`StreamMode.WRITE` that has
-   similar API of :class:`StreamWriter`. It can also be used as an async context manager.
-
-   *limit* determines the buffer size limit used by the returned :class:`Stream`
-   instance. By default the limit is set to 64 KiB.
-
-   .. versionadded:: 3.8
 
 .. rubric:: Unix Sockets
 
-.. function:: connect_unix(path=None, *, limit=2**16, ssl=None, \
-                           sock=None, server_hostname=None, \
-                           ssl_handshake_timeout=None)
-
-   Establish a Unix socket connection to socket with *path* address and
-   return an awaitable :class:`Stream` object of the mode :attr:`StreamMode.READWRITE`
-   that can be used as a reader and a writer.
-
-   *limit* determines the buffer size limit used by the returned :class:`Stream`
-   instance. By default the *limit* is set to 64 KiB.
-
-   The rest of the arguments are passed directly to :meth:`loop.create_unix_connection`.
-
-   The function can be used with ``await`` to get a connected stream::
-
-       stream = await asyncio.connect_unix('/tmp/example.sock')
-
-   The function can also be used as an async context manager::
-
-       async with asyncio.connect_unix('/tmp/example.sock') as stream:
-           ...
-
-   .. availability:: Unix.
-
-   .. versionadded:: 3.8
-
 .. coroutinefunction:: open_unix_connection(path=None, \*, loop=None, \
                         limit=None, ssl=None, sock=None, \
                         server_hostname=None, ssl_handshake_timeout=None)
@@ -199,10 +126,6 @@ and work with streams:
 
       The *path* parameter can now be a :term:`path-like object`
 
-   .. deprecated-removed:: 3.8 3.10
-
-      ``open_unix_connection()`` is deprecated if favor of :func:`connect_unix`.
-
 
 .. coroutinefunction:: start_unix_server(client_connected_cb, path=None, \
                           \*, loop=None, limit=None, sock=None, \
@@ -225,349 +148,6 @@ and work with streams:
 
       The *path* parameter can now be a :term:`path-like object`.
 
-   .. deprecated-removed:: 3.8 3.10
-
-      ``start_unix_server()`` is deprecated in favor of :class:`UnixStreamServer`.
-
-
----------
-
-StreamServer
-============
-
-.. class:: StreamServer(client_connected_cb, /, host=None, port=None, *, \
-                        limit=2**16, family=socket.AF_UNSPEC, \
-                        flags=socket.AI_PASSIVE, sock=None, backlog=100, \
-                        ssl=None, reuse_address=None, reuse_port=None, \
-                        ssl_handshake_timeout=None, shutdown_timeout=60)
-
-   The *client_connected_cb* callback is called whenever a new client
-   connection is established.  It receives a :class:`Stream` object of the
-   mode :attr:`StreamMode.READWRITE`.
-
-   *client_connected_cb* can be a plain callable or a
-   :ref:`coroutine function <coroutine>`; if it is a coroutine function,
-   it will be automatically scheduled as a :class:`Task`.
-
-   *limit* determines the buffer size limit used by the
-   returned :class:`Stream` instance.  By default the *limit*
-   is set to 64 KiB.
-
-   The rest of the arguments are passed directly to
-   :meth:`loop.create_server`.
-
-   .. coroutinemethod:: start_serving()
-
-      Binds to the given host and port to start the server.
-
-   .. coroutinemethod:: serve_forever()
-
-      Start accepting connections until the coroutine is cancelled.
-      Cancellation of ``serve_forever`` task causes the server
-      to be closed.
-
-      This method can be called if the server is already accepting
-      connections.  Only one ``serve_forever`` task can exist per
-      one *Server* object.
-
-   .. method:: is_serving()
-
-      Returns ``True`` if the server is bound and currently serving.
-
-   .. method:: bind()
-
-      Bind the server to the given *host* and *port*. This method is
-      automatically called during ``__aenter__`` when :class:`StreamServer` is
-      used as an async context manager.
-
-   .. method:: is_bound()
-
-      Return ``True`` if the server is bound.
-
-   .. coroutinemethod:: abort()
-
-      Closes the connection and cancels all pending tasks.
-
-   .. coroutinemethod:: close()
-
-      Closes the connection. This method is automatically called during
-      ``__aexit__`` when :class:`StreamServer` is used as an async context
-      manager.
-
-   .. attribute:: sockets
-
-      Returns a tuple of socket objects the server is bound to.
-
-   .. versionadded:: 3.8
-
-
-UnixStreamServer
-================
-
-.. class:: UnixStreamServer(client_connected_cb, /, path=None, *, \
-                            limit=2**16, sock=None, backlog=100, \
-                            ssl=None, ssl_handshake_timeout=None, shutdown_timeout=60)
-
-   The *client_connected_cb* callback is called whenever a new client
-   connection is established.  It receives a :class:`Stream` object of the
-   mode :attr:`StreamMode.READWRITE`.
-
-   *client_connected_cb* can be a plain callable or a
-   :ref:`coroutine function <coroutine>`; if it is a coroutine function,
-   it will be automatically scheduled as a :class:`Task`.
-
-   *limit* determines the buffer size limit used by the
-   returned :class:`Stream` instance.  By default the *limit*
-   is set to 64 KiB.
-
-   The rest of the arguments are passed directly to
-   :meth:`loop.create_unix_server`.
-
-   .. coroutinemethod:: start_serving()
-
-      Binds to the given host and port to start the server.
-
-   .. method:: is_serving()
-
-      Returns ``True`` if the server is bound and currently serving.
-
-   .. method:: bind()
-
-      Bind the server to the given *host* and *port*. This method is
-      automatically called during ``__aenter__`` when :class:`UnixStreamServer` is
-      used as an async context manager.
-
-   .. method:: is_bound()
-
-      Return ``True`` if the server is bound.
-
-   .. coroutinemethod:: abort()
-
-      Closes the connection and cancels all pending tasks.
-
-   .. coroutinemethod:: close()
-
-      Closes the connection. This method is automatically called during
-      ``__aexit__`` when :class:`UnixStreamServer` is used as an async context
-      manager.
-
-   .. attribute:: sockets
-
-      Returns a tuple of socket objects the server is bound to.
-
-   .. availability:: Unix.
-
-   .. versionadded:: 3.8
-
-Stream
-======
-
-.. class:: Stream
-
-   Represents a Stream object that provides APIs to read and write data
-   to the IO stream . It includes the API provided by :class:`StreamReader`
-   and :class:`StreamWriter`. It can also be used as :term:`asynchronous iterator`
-   where :meth:`readline` is used. It raises :exc:`StopAsyncIteration` when
-   :meth:`readline` returns empty data.
-
-   Do not instantiate *Stream* objects directly; use API like :func:`connect`
-   and :class:`StreamServer` instead.
-
-   .. versionadded:: 3.8
-
-   .. attribute:: mode
-
-      Returns the mode of the stream which is a :class:`StreamMode` value. It could
-      be one of the below:
-
-      * :attr:`StreamMode.READ` - Connection can receive data.
-      * :attr:`StreamMode.WRITE` - Connection can send data.
-      * :attr:`StreamMode.READWRITE` - Connection can send and receive data.
-
-   .. coroutinemethod:: abort()
-
-      Aborts the connection immediately, without waiting for the send buffer to drain.
-
-   .. method:: at_eof()
-
-      Return ``True`` if the buffer is empty.
-
-   .. method:: can_write_eof()
-
-      Return *True* if the underlying transport supports
-      the :meth:`write_eof` method, *False* otherwise.
-
-   .. method:: close()
-
-      The method closes the stream and the underlying socket.
-
-      It is possible to directly await on the `close()` method::
-
-         await stream.close()
-
-      The ``await`` pauses the current coroutine until the stream and the underlying
-      socket are closed (and SSL shutdown is performed for a secure connection).
-
-   .. coroutinemethod:: drain()
-
-      Wait until it is appropriate to resume writing to the stream.
-      Example::
-
-          stream.write(data)
-          await stream.drain()
-
-      This is a flow control method that interacts with the underlying
-      IO write buffer.  When the size of the buffer reaches
-      the high watermark, *drain()* blocks until the size of the
-      buffer is drained down to the low watermark and writing can
-      be resumed.  When there is nothing to wait for, the :meth:`drain`
-      returns immediately.
-
-      .. deprecated:: 3.8
-
-      It is recommended to directly await on the `write()` method instead::
-
-         await stream.write(data)
-
-   .. method:: get_extra_info(name, default=None)
-
-      Access optional transport information; see
-      :meth:`BaseTransport.get_extra_info` for details.
-
-   .. method:: is_closing()
-
-      Return ``True`` if the stream is closed or in the process of
-      being closed.
-
-   .. coroutinemethod:: read(n=-1)
-
-      Read up to *n* bytes.  If *n* is not provided, or set to ``-1``,
-      read until EOF and return all read bytes.
-
-      If EOF was received and the internal buffer is empty,
-      return an empty ``bytes`` object.
-
-   .. coroutinemethod:: readexactly(n)
-
-      Read exactly *n* bytes.
-
-      Raise an :exc:`IncompleteReadError` if EOF is reached before *n*
-      can be read.  Use the :attr:`IncompleteReadError.partial`
-      attribute to get the partially read data.
-
-   .. coroutinemethod:: readline()
-
-      Read one line, where "line" is a sequence of bytes
-      ending with ``\n``.
-
-      If EOF is received and ``\n`` was not found, the method
-      returns partially read data.
-
-      If EOF is received and the internal buffer is empty,
-      return an empty ``bytes`` object.
-
-   .. coroutinemethod:: readuntil(separator=b'\\n')
-
-      Read data from the stream until *separator* is found.
-
-      On success, the data and separator will be removed from the
-      internal buffer (consumed). Returned data will include the
-      separator at the end.
-
-      If the amount of data read exceeds the configured stream limit, a
-      :exc:`LimitOverrunError` exception is raised, and the data
-      is left in the internal buffer and can be read again.
-
-      If EOF is reached before the complete separator is found,
-      an :exc:`IncompleteReadError` exception is raised, and the internal
-      buffer is reset.  The :attr:`IncompleteReadError.partial` attribute
-      may contain a portion of the separator.
-
-   .. coroutinemethod:: sendfile(file, offset=0, count=None, *, fallback=True)
-
-      Sends a *file* over the stream using an optimized syscall if available.
-
-      For other parameters meaning please see :meth:`AbstractEventloop.sendfile`.
-
-   .. coroutinemethod:: start_tls(sslcontext, *, server_hostname=None, \
-                                  ssl_handshake_timeout=None)
-
-      Upgrades the existing transport-based connection to TLS.
-
-      For other parameters meaning please see :meth:`AbstractEventloop.start_tls`.
-
-   .. coroutinemethod:: wait_closed()
-
-      Wait until the stream is closed.
-
-      Should be called after :meth:`close` to wait until the underlying
-      connection is closed.
-
-   .. coroutinemethod:: write(data)
-
-      Write *data* to the underlying socket; wait until the data is sent, e.g.::
-
-         await stream.write(data)
-
-   .. method:: write(data)
-
-      The method attempts to write the *data* to the underlying socket immediately.
-      If that fails, the data is queued in an internal write buffer until it can be
-      sent. :meth:`drain` can be used to flush the underlying buffer once writing is
-      available::
-
-         stream.write(data)
-         await stream.drain()
-
-      .. deprecated:: 3.8
-
-      It is recommended to directly await on the `write()` method instead::
-
-          await stream.write(data)
-
-   .. method:: writelines(data)
-
-      The method writes a list (or any iterable) of bytes to the underlying socket
-      immediately.
-      If that fails, the data is queued in an internal write buffer until it can be
-      sent.
-
-      It is possible to directly await on the `writelines()` method::
-
-         await stream.writelines(lines)
-
-      The ``await`` pauses the current coroutine until the data is written to the
-      socket.
-
-   .. method:: write_eof()
-
-      Close the write end of the stream after the buffered write
-      data is flushed.
-
-
-StreamMode
-==========
-
-.. class:: StreamMode
-
-   A subclass of :class:`enum.Flag` that defines a set of values that can be
-   used to determine the ``mode`` of :class:`Stream` objects.
-
-   .. data:: READ
-
-   The stream object is readable and provides the API of :class:`StreamReader`.
-
-   .. data:: WRITE
-
-   The stream object is writeable and provides the API of :class:`StreamWriter`.
-
-   .. data:: READWRITE
-
-   The stream object is readable and writeable and provides the API of both
-   :class:`StreamReader` and :class:`StreamWriter`.
-
-  .. versionadded:: 3.8
-
 
 StreamReader
 ============
@@ -629,7 +209,8 @@ StreamReader
 
    .. method:: at_eof()
 
-      Return ``True`` if the buffer is empty.
+      Return ``True`` if the buffer is empty and :meth:`feed_eof`
+      was called.
 
 
 StreamWriter
@@ -650,22 +231,11 @@ StreamWriter
       If that fails, the data is queued in an internal write buffer until it can be
       sent.
 
-      Starting with Python 3.8, it is possible to directly await on the `write()`
+      The method should be used along with the ``drain()`` method::
-      method::
-
-         await stream.write(data)
-
-      The ``await`` pauses the current coroutine until the data is written to the
-      socket.
-
-      Below is an equivalent code that works with Python <= 3.7::
 
          stream.write(data)
          await stream.drain()
 
-      .. versionchanged:: 3.8
-         Support ``await stream.write(...)`` syntax.
-
    .. method:: writelines(data)
 
       The method writes a list (or any iterable) of bytes to the underlying socket
@@ -673,42 +243,20 @@ StreamWriter
       If that fails, the data is queued in an internal write buffer until it can be
       sent.
 
-      Starting with Python 3.8, it is possible to directly await on the `writelines()`
+      The method should be used along with the ``drain()`` method::
-      method::
-
-         await stream.writelines(lines)
-
-      The ``await`` pauses the current coroutine until the data is written to the
-      socket.
-
-      Below is an equivalent code that works with Python <= 3.7::
 
          stream.writelines(lines)
          await stream.drain()
 
-      .. versionchanged:: 3.8
-         Support ``await stream.writelines()`` syntax.
-
    .. method:: close()
 
       The method closes the stream and the underlying socket.
 
-      Starting with Python 3.8, it is possible to directly await on the `close()`
+      The method should be used along with the ``wait_closed()`` method::
-      method::
-
-         await stream.close()
-
-      The ``await`` pauses the current coroutine until the stream and the underlying
-      socket are closed (and SSL shutdown is performed for a secure connection).
-
-      Below is an equivalent code that works with Python <= 3.7::
 
          stream.close()
          await stream.wait_closed()
 
-      .. versionchanged:: 3.8
-         Support ``await stream.close()`` syntax.
-
    .. method:: can_write_eof()
 
       Return *True* if the underlying transport supports
@@ -768,17 +316,22 @@ Examples
 TCP echo client using streams
 -----------------------------
 
-TCP echo client using the :func:`asyncio.connect` function::
+TCP echo client using the :func:`asyncio.open_connection` function::
 
     import asyncio
 
     async def tcp_echo_client(message):
-        async with asyncio.connect('127.0.0.1', 8888) as stream:
+        reader, writer = await asyncio.open_connection(
-            print(f'Send: {message!r}')
+            '127.0.0.1', 8888)
-            await stream.write(message.encode())
 
-            data = await stream.read(100)
+        print(f'Send: {message!r}')
-            print(f'Received: {data.decode()!r}')
+        writer.write(message.encode())
+
+        data = await reader.read(100)
+        print(f'Received: {data.decode()!r}')
+
+        print('Close the connection')
+        writer.close()
 
     asyncio.run(tcp_echo_client('Hello World!'))
 
@@ -794,28 +347,32 @@ TCP echo client using the :func:`asyncio.connect` function::
 TCP echo server using streams
 -----------------------------
 
-TCP echo server using the :class:`asyncio.StreamServer` class::
+TCP echo server using the :func:`asyncio.start_server` function::
 
     import asyncio
 
-    async def handle_echo(stream):
+    async def handle_echo(reader, writer):
-        data = await stream.read(100)
+        data = await reader.read(100)
         message = data.decode()
-        addr = stream.get_extra_info('peername')
+        addr = writer.get_extra_info('peername')
 
         print(f"Received {message!r} from {addr!r}")
 
         print(f"Send: {message!r}")
-        await stream.write(data)
+        writer.write(data)
+        await writer.drain()
 
         print("Close the connection")
-        await stream.close()
+        writer.close()
 
     async def main():
-        async with asyncio.StreamServer(
+        server = await asyncio.start_server(
-                handle_echo, '127.0.0.1', 8888) as server:
+            handle_echo, '127.0.0.1', 8888)
-            addr = server.sockets[0].getsockname()
+
-            print(f'Serving on {addr}')
+        addr = server.sockets[0].getsockname()
+        print(f'Serving on {addr}')
+
+        async with server:
             await server.serve_forever()
 
     asyncio.run(main())
@@ -839,9 +396,11 @@ Simple example querying HTTP headers of the URL passed on the command line::
     async def print_http_headers(url):
         url = urllib.parse.urlsplit(url)
         if url.scheme == 'https':
-            stream = await asyncio.connect(url.hostname, 443, ssl=True)
+            reader, writer = await asyncio.open_connection(
+                url.hostname, 443, ssl=True)
         else:
-            stream = await asyncio.connect(url.hostname, 80)
+            reader, writer = await asyncio.open_connection(
+                url.hostname, 80)
 
         query = (
             f"HEAD {url.path or '/'} HTTP/1.0\r\n"
@@ -849,14 +408,18 @@ Simple example querying HTTP headers of the URL passed on the command line::
             f"\r\n"
         )
 
-        stream.write(query.encode('latin-1'))
+        writer.write(query.encode('latin-1'))
-        while (line := await stream.readline()):
+        while True:
+            line = await reader.readline()
+            if not line:
+                break
+
             line = line.decode('latin1').rstrip()
             if line:
                 print(f'HTTP header> {line}')
 
         # Ignore the body, close the socket
-        await stream.close()
+        writer.close()
 
     url = sys.argv[1]
     asyncio.run(print_http_headers(url))
@@ -877,7 +440,7 @@ Register an open socket to wait for data using streams
 ------------------------------------------------------
 
 Coroutine waiting until a socket receives data using the
-:func:`asyncio.connect` function::
+:func:`open_connection` function::
 
     import asyncio
     import socket
@@ -891,15 +454,17 @@ Coroutine waiting until a socket receives data using the
         rsock, wsock = socket.socketpair()
 
         # Register the open socket to wait for data.
-        async with asyncio.connect(sock=rsock) as stream:
+        reader, writer = await asyncio.open_connection(sock=rsock)
-            # Simulate the reception of data from the network
-            loop.call_soon(wsock.send, 'abc'.encode())
 
-            # Wait for data
+        # Simulate the reception of data from the network
-            data = await stream.read(100)
+        loop.call_soon(wsock.send, 'abc'.encode())
 
-            # Got data, we are done: close the socket
+        # Wait for data
-            print("Received:", data.decode())
+        data = await reader.read(100)
+
+        # Got data, we are done: close the socket
+        print("Received:", data.decode())
+        writer.close()
 
         # Close the second socket
         wsock.close()

Lib/asyncio/__init__.py

@@ -3,7 +3,6 @@
 # flake8: noqa
 
 import sys
-import warnings
 
 # This relies on each of the submodules having an __all__ variable.
 from .base_events import *
@@ -44,40 +43,3 @@ 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

@@ -19,16 +19,14 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
                                protocols.SubprocessProtocol):
     """Like StreamReaderProtocol, but for a subprocess."""
 
-    def __init__(self, limit, loop, *, _asyncio_internal=False):
+    def __init__(self, limit, loop):
-        super().__init__(loop=loop, _asyncio_internal=_asyncio_internal)
+        super().__init__(loop=loop)
         self._limit = limit
         self.stdin = self.stdout = self.stderr = None
         self._transport = None
         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__]
@@ -42,35 +40,27 @@ 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.Stream(mode=streams.StreamMode.READ,
+            self.stdout = streams.StreamReader(limit=self._limit,
-                                         transport=stdout_transport,
+                                               loop=self._loop)
-                                         protocol=self,
+            self.stdout.set_transport(stdout_transport)
-                                         limit=self._limit,
-                                         loop=self._loop,
-                                         _asyncio_internal=True)
-            self.stdout._set_transport(stdout_transport)
             self._pipe_fds.append(1)
 
         stderr_transport = transport.get_pipe_transport(2)
         if stderr_transport is not None:
-            self.stderr = streams.Stream(mode=streams.StreamMode.READ,
+            self.stderr = streams.StreamReader(limit=self._limit,
-                                         transport=stderr_transport,
+                                               loop=self._loop)
-                                         protocol=self,
+            self.stderr.set_transport(stderr_transport)
-                                         limit=self._limit,
-                                         loop=self._loop,
-                                         _asyncio_internal=True)
-            self.stderr._set_transport(stderr_transport)
             self._pipe_fds.append(2)
 
         stdin_transport = transport.get_pipe_transport(0)
         if stdin_transport is not None:
-            self.stdin = streams.Stream(mode=streams.StreamMode.WRITE,
+            self.stdin = streams.StreamWriter(stdin_transport,
-                                        transport=stdin_transport,
+                                              protocol=self,
-                                        protocol=self,
+                                              reader=None,
-                                        loop=self._loop,
+                                              loop=self._loop)
-                                        _asyncio_internal=True)
 
     def pipe_data_received(self, fd, data):
         if fd == 1:
@@ -80,7 +70,7 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
         else:
             reader = None
         if reader is not None:
-            reader._feed_data(data)
+            reader.feed_data(data)
 
     def pipe_connection_lost(self, fd, exc):
         if fd == 0:
@@ -101,9 +91,9 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
             reader = None
         if reader is not None:
             if exc is None:
-                reader._feed_eof()
+                reader.feed_eof()
             else:
-                reader._set_exception(exc)
+                reader.set_exception(exc)
 
         if fd in self._pipe_fds:
             self._pipe_fds.remove(fd)
@@ -121,20 +111,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:
-    def __init__(self, transport, protocol, loop, *, _asyncio_internal=False):
+    def __init__(self, transport, protocol, loop):
-        if not _asyncio_internal:
-            warnings.warn(f"{self.__class__} should be instantiated "
-                          "by asyncio internals only, "
-                          "please avoid its creation from user code",
-                          DeprecationWarning)
-
         self._transport = transport
         self._protocol = protocol
         self._loop = loop
@@ -232,13 +212,12 @@ async def create_subprocess_shell(cmd, stdin=None, stdout=None, stderr=None,
         )
 
     protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
-                                                        loop=loop,
+                                                        loop=loop)
-                                                        _asyncio_internal=True)
     transport, protocol = await loop.subprocess_shell(
         protocol_factory,
         cmd, stdin=stdin, stdout=stdout,
         stderr=stderr, **kwds)
-    return Process(transport, protocol, loop, _asyncio_internal=True)
+    return Process(transport, protocol, loop)
 
 
 async def create_subprocess_exec(program, *args, stdin=None, stdout=None,
@@ -253,11 +232,10 @@ async def create_subprocess_exec(program, *args, stdin=None, stdout=None,
                       stacklevel=2
         )
     protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
-                                                        loop=loop,
+                                                        loop=loop)
-                                                        _asyncio_internal=True)
     transport, protocol = await loop.subprocess_exec(
         protocol_factory,
         program, *args,
         stdin=stdin, stdout=stdout,
         stderr=stderr, **kwds)
-    return Process(transport, protocol, loop, _asyncio_internal=True)
+    return Process(transport, protocol, loop)

Lib/test/test_asyncio/test_buffered_proto.py

@@ -58,10 +58,9 @@ class BaseTestBufferedProtocol(func_tests.FunctionalTestCaseMixin):
             writer.close()
             await writer.wait_closed()
 
-        with self.assertWarns(DeprecationWarning):
+        srv = self.loop.run_until_complete(
-            srv = self.loop.run_until_complete(
+            asyncio.start_server(
-                asyncio.start_server(
+                on_server_client, '127.0.0.1', 0))
-                    on_server_client, '127.0.0.1', 0))
 
         addr = srv.sockets[0].getsockname()
         self.loop.run_until_complete(

Lib/test/test_asyncio/test_pep492.py

@@ -95,11 +95,9 @@ class StreamReaderTests(BaseTest):
     def test_readline(self):
         DATA = b'line1\nline2\nline3'
 
-        stream = asyncio.Stream(mode=asyncio.StreamMode.READ,
+        stream = asyncio.StreamReader(loop=self.loop)
-                                loop=self.loop,
+        stream.feed_data(DATA)
-                                _asyncio_internal=True)
+        stream.feed_eof()
-        stream._feed_data(DATA)
-        stream._feed_eof()
 
         async def reader():
             data = []

Lib/test/test_asyncio/test_subprocess.py

@@ -582,18 +582,6 @@ class SubprocessMixin:
 
         self.loop.run_until_complete(execute())
 
-    def test_subprocess_protocol_create_warning(self):
-        with self.assertWarns(DeprecationWarning):
-            subprocess.SubprocessStreamProtocol(limit=10, loop=self.loop)
-
-    def test_process_create_warning(self):
-        proto = subprocess.SubprocessStreamProtocol(limit=10, loop=self.loop,
-                                                    _asyncio_internal=True)
-        transp = mock.Mock()
-
-        with self.assertWarns(DeprecationWarning):
-            subprocess.Process(transp, proto, loop=self.loop)
-
     def test_create_subprocess_exec_text_mode_fails(self):
         async def execute():
             with self.assertRaises(ValueError):

Lib/test/test_asyncio/test_windows_events.py

@@ -15,7 +15,6 @@ import _winapi
 
 import asyncio
 from asyncio import windows_events
-from asyncio.streams import _StreamProtocol
 from test.test_asyncio import utils as test_utils
 
 
@@ -118,16 +117,14 @@ class ProactorTests(test_utils.TestCase):
 
         clients = []
         for i in range(5):
-            stream = asyncio.Stream(mode=asyncio.StreamMode.READ,
+            stream_reader = asyncio.StreamReader(loop=self.loop)
-                                    loop=self.loop, _asyncio_internal=True)
+            protocol = asyncio.StreamReaderProtocol(stream_reader,
-            protocol = _StreamProtocol(stream,
+                                                    loop=self.loop)
-                                       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, trans))
+            clients.append((stream_reader, trans))
 
         for i, (r, w) in enumerate(clients):
             w.write('lower-{}\n'.format(i).encode())
@@ -136,7 +133,6 @@ 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-09-30-00-15-27.bpo-38242.uPIyAc.rst

@@ -0,0 +1 @@
+Revert the new asyncio Streams API