2 from _typeshed import FileDescriptorLike
3 from socket import socket
4 from typing import Any, Dict, Tuple, overload
6 # cyclic dependence with asynchat
7 _maptype = Dict[int, Any]
10 socket_map: _maptype # undocumented
12 class ExitNow(Exception): ...
14 def read(obj: Any) -> None: ...
15 def write(obj: Any) -> None: ...
16 def readwrite(obj: Any, flags: int) -> None: ...
17 def poll(timeout: float = ..., map: _maptype | None = ...) -> None: ...
18 def poll2(timeout: float = ..., map: _maptype | None = ...) -> None: ...
22 def loop(timeout: float = ..., use_poll: bool = ..., map: _maptype | None = ..., count: int | None = ...) -> None: ...
24 # Not really subclass of socket.socket; it's only delegation.
25 # It is not covariant to it.
33 ignore_log_types: frozenset[str]
34 socket: _socket | None
35 def __init__(self, sock: _socket | None = ..., map: _maptype | None = ...) -> None: ...
36 def add_channel(self, map: _maptype | None = ...) -> None: ...
37 def del_channel(self, map: _maptype | None = ...) -> None: ...
38 def create_socket(self, family: int = ..., type: int = ...) -> None: ...
39 def set_socket(self, sock: _socket, map: _maptype | None = ...) -> None: ...
40 def set_reuse_addr(self) -> None: ...
41 def readable(self) -> bool: ...
42 def writable(self) -> bool: ...
43 def listen(self, num: int) -> None: ...
44 def bind(self, addr: Tuple[Any, ...] | str) -> None: ...
45 def connect(self, address: Tuple[Any, ...] | str) -> None: ...
46 def accept(self) -> tuple[_socket, Any] | None: ...
47 def send(self, data: bytes) -> int: ...
48 def recv(self, buffer_size: int) -> bytes: ...
49 def close(self) -> None: ...
50 def log(self, message: Any) -> None: ...
51 def log_info(self, message: Any, type: str = ...) -> None: ...
52 def handle_read_event(self) -> None: ...
53 def handle_connect_event(self) -> None: ...
54 def handle_write_event(self) -> None: ...
55 def handle_expt_event(self) -> None: ...
56 def handle_error(self) -> None: ...
57 def handle_expt(self) -> None: ...
58 def handle_read(self) -> None: ...
59 def handle_write(self) -> None: ...
60 def handle_connect(self) -> None: ...
61 def handle_accept(self) -> None: ...
62 def handle_close(self) -> None: ...
64 class dispatcher_with_send(dispatcher):
65 def __init__(self, sock: socket | None = ..., map: _maptype | None = ...) -> None: ...
66 def initiate_send(self) -> None: ...
67 def handle_write(self) -> None: ...
68 # incompatible signature:
69 # def send(self, data: bytes) -> int | None: ...
71 def compact_traceback() -> tuple[tuple[str, str, str], type, type, str]: ...
72 def close_all(map: _maptype | None = ..., ignore_all: bool = ...) -> None: ...
74 if sys.platform != "win32":
77 def __init__(self, fd: int) -> None: ...
78 def recv(self, bufsize: int, flags: int = ...) -> bytes: ...
79 def send(self, data: bytes, flags: int = ...) -> int: ...
81 def getsockopt(self, level: int, optname: int, buflen: None = ...) -> int: ...
83 def getsockopt(self, level: int, optname: int, buflen: int) -> bytes: ...
84 def read(self, bufsize: int, flags: int = ...) -> bytes: ...
85 def write(self, data: bytes, flags: int = ...) -> int: ...
86 def close(self) -> None: ...
87 def fileno(self) -> int: ...
88 class file_dispatcher(dispatcher):
89 def __init__(self, fd: FileDescriptorLike, map: _maptype | None = ...) -> None: ...
90 def set_file(self, fd: int) -> None: ...