1"""Helper methods to tune a TCP connection""" 2 3import asyncio 4import socket 5from contextlib import suppress 6from typing import Optional # noqa 7 8__all__ = ("tcp_keepalive", "tcp_nodelay") 9 10 11if hasattr(socket, "SO_KEEPALIVE"): 12 13 def tcp_keepalive(transport: asyncio.Transport) -> None: 14 sock = transport.get_extra_info("socket") 15 if sock is not None: 16 sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) 17 18 19else: 20 21 def tcp_keepalive(transport: asyncio.Transport) -> None: # pragma: no cover 22 pass 23 24 25def tcp_nodelay(transport: asyncio.Transport, value: bool) -> None: 26 sock = transport.get_extra_info("socket") 27 28 if sock is None: 29 return 30 31 if sock.family not in (socket.AF_INET, socket.AF_INET6): 32 return 33 34 value = bool(value) 35 36 # socket may be closed already, on windows OSError get raised 37 with suppress(OSError): 38 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, value) 39