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