Added connect_accepted_socket API

HISTORY.rst

@@ -10,7 +10,8 @@ aiofastnet Release History
 
 * Harden logic against exceptions in SSLTransport constructor
 * Added create_datagram_endpoint
-* Fixed Protocol.connection_lost may not be called if failure and abort happenes during start_tls
+* Added connect_accepted_socket
+* Fixed Protocol.connection_lost may not be called if failure and abort happens during start_tls
 
 0.21.0
 ------------------

README.md

@@ -69,6 +69,7 @@ Source: [examples/benchmark_threaded.py](https://github.com/tarasko/aiofastnet/b
 `aiofastnet` provides drop-in, highly efficient C/Cython replacements for asyncio's:
 
 - [`loop.create_connection()`](https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.create_connection)
+- [`loop.connect_accepted_socket()`](https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.connect_accepted_socket)
 - [`loop.open_connection()`](https://docs.python.org/3/library/asyncio-stream.html#asyncio.open_connection)
 - [`loop.create_unix_connection()`](https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.create_unix_connection)
 - [`loop.open_unix_connection()`](https://docs.python.org/3/library/asyncio-stream.html#asyncio.open_unix_connection)

aiofastnet/__init__.py

@@ -1,5 +1,6 @@
 import socket
 
+from .api_connect_accepted_socket import connect_accepted_socket
 from .api_create_connection import create_connection
 from .api_create_datagram_endpoint import create_datagram_endpoint
 from .api_create_server import create_server
@@ -20,6 +21,7 @@
     'Protocol',
     'Transport',
     'aiofn_is_buffered_protocol',
+    'connect_accepted_socket',
     'create_connection',
     'create_datagram_endpoint',
     'create_server',

aiofastnet/__init__.pyi

@@ -63,6 +63,18 @@ async def create_connection(
     all_errors: bool = ...,
 ) -> tuple[asyncio.Transport, _ProtocolT]: ...
 
+async def connect_accepted_socket(
+    loop: asyncio.AbstractEventLoop,
+    protocol_factory: Callable[[], _ProtocolT],
+    sock: socket.socket,
+    *,
+    ssl: bool | ssl.SSLContext | None = ...,
+    ssl_handshake_timeout: float | None = ...,
+    ssl_shutdown_timeout: float | None = ...,
+    ssl_incoming_bio_size: int | None = ...,
+    ssl_outgoing_bio_size: int | None = ...,
+) -> tuple[asyncio.Transport, _ProtocolT]: ...
+
 async def create_datagram_endpoint(
     loop: asyncio.AbstractEventLoop,
     protocol_factory: Callable[[], _DatagramProtocolT],

aiofastnet/api_connect_accepted_socket.py

@@ -0,0 +1,46 @@
+# Portions of this file are derived from CPython's asyncio sources
+# (notably asyncio.base_events and asyncio.selector_events).
+# Copyright (c) Python Software Foundation.
+# Licensed under the Python Software Foundation License Version 2.
+# See LICENSES/PSF-2.0.txt and THIRD_PARTY_NOTICES for details.
+
+import socket
+
+from .api_utils import _check_ssl_socket, _create_connection_transport, _logger, _validate_bio_size, _validate_ssl_timeout
+
+
+async def connect_accepted_socket(
+    loop,
+    protocol_factory,
+    sock,
+    *,
+    ssl=None,
+    ssl_handshake_timeout=None,
+    ssl_shutdown_timeout=None,
+    ssl_incoming_bio_size=None,
+    ssl_outgoing_bio_size=None,
+):
+    if sock.type != socket.SOCK_STREAM:
+        raise ValueError(f"A Stream Socket was expected, got {sock!r}")
+
+    ssl_handshake_timeout = _validate_ssl_timeout("ssl_handshake_timeout", ssl_handshake_timeout, ssl)
+    ssl_shutdown_timeout = _validate_ssl_timeout("ssl_shutdown_timeout", ssl_shutdown_timeout, ssl)
+    ssl_incoming_bio_size = _validate_bio_size("ssl_incoming_bio_size", ssl_incoming_bio_size, ssl)
+    ssl_outgoing_bio_size = _validate_bio_size("ssl_outgoing_bio_size", ssl_outgoing_bio_size, ssl)
+
+    _check_ssl_socket(sock)
+
+    transport, protocol = await _create_connection_transport(
+        loop, sock, protocol_factory, ssl, "",
+        server_side=True,
+        ssl_handshake_timeout=ssl_handshake_timeout,
+        ssl_shutdown_timeout=ssl_shutdown_timeout,
+        ssl_incoming_bio_size=ssl_incoming_bio_size,
+        ssl_outgoing_bio_size=ssl_outgoing_bio_size,
+    )
+    if loop.get_debug():
+        # Get the socket from the transport because SSL transport closes
+        # the old socket and creates a new SSL socket
+        sock = transport.get_extra_info("socket")
+        _logger.debug("%r handled: (%r, %r)", sock, transport, protocol)
+    return transport, protocol

tests/test_smoke.py

@@ -13,7 +13,33 @@
 
 import aiofastnet
 from aiofastnet.transport import Protocol, SocketTransport, Transport
-from tests.utils import UDP_MAX_PAYLOAD_SIZE, AsyncClient, SomeException, TestClient, TestServer, _logger, make_test_ssl_contexts, sendfile, start_tls
+from tests.utils import (
+    UDP_MAX_PAYLOAD_SIZE,
+    AsyncClient,
+    EchoServerProtocol,
+    SocketPair,
+    SomeException,
+    TestClient,
+    TestServer,
+    _logger,
+    make_test_ssl_contexts,
+    sendfile,
+    start_tls,
+)
+
+
+async def test_echo_socketpair(conn_type_plus_udp):
+    msg_size = 1024
+    payload = b"x" * msg_size
+
+    async with SocketPair(ct=conn_type_plus_udp,
+                          server_protocol_factory=EchoServerProtocol,
+                          client_server_hostname="127.0.0.1") as (_server, client):
+        client.write(payload)
+        echoed = await client.readn(msg_size)
+        assert echoed == payload
+        client.close()
+        await client.wait_closed()
 
 
 @pytest.mark.parametrize("msg_size", [1, 2, 3, 4, 5, 6, 7, 8, 29, 64, 256 * 1024, 6 * 1024 * 1024])
@@ -35,6 +61,18 @@ async def test_echo(all_loops, msg_size, conn_type_plus_udp, buffered_protocol):
             await client.wait_closed()
 
 
+@pytest.mark.parametrize("msg_size", [1, 32, 64, 256 * 1024, 6 * 1024 * 1024, 20 * 1024 * 1024])
+@pytest.mark.parametrize("num_lines", [1, 32, 4000])
+async def test_echo_writelines(all_loops, msg_size, num_lines, conn_type, buffered_protocol):
+    payload = b"x" * msg_size
+
+    async with TestServer(ct=conn_type, is_buffered=buffered_protocol) as server:
+        async with TestClient(server, ct=conn_type, is_buffered=buffered_protocol) as client:
+            client.write_in_lines(payload, num_lines)
+            echoed = await client.readn(msg_size, 4.0)
+            assert echoed == payload
+
+
 async def test_ktls_enabled(ktls_conn_type):
     async with TestServer(ct=ktls_conn_type) as server:
         async with TestClient(server, ct=ktls_conn_type) as client:
@@ -78,18 +116,6 @@ async def test_ssl_membio_enabled(selector_loop, ssl_conn_type):
             assert server_client.transport.get_extra_info("ssl_outgoing_use_membio") is expected
 
 
-@pytest.mark.parametrize("msg_size", [1, 32, 64, 256 * 1024, 6 * 1024 * 1024, 20 * 1024 * 1024])
-@pytest.mark.parametrize("num_lines", [1, 32, 4000])
-async def test_echo_writelines(all_loops, msg_size, num_lines, conn_type, buffered_protocol):
-    payload = b"x" * msg_size
-
-    async with TestServer(ct=conn_type, is_buffered=buffered_protocol) as server:
-        async with TestClient(server, ct=conn_type, is_buffered=buffered_protocol) as client:
-            client.write_in_lines(payload, num_lines)
-            echoed = await client.readn(msg_size, 4.0)
-            assert echoed == payload
-
-
 async def test_write_huge_close(all_loops, conn_type):
     if os.name == 'nt' and isinstance(asyncio.get_running_loop(), asyncio.ProactorEventLoop) and sys.version_info < (3, 11):
         pytest.skip("ProactorEventLoop in 3.9 and 3.10 had issues with connection closing")