import concurrent.futures import socket import ssl from tornado_py3.concurrent import dummy_executor as dummy_executor, run_on_executor as run_on_executor from tornado_py3.ioloop import IOLoop as IOLoop from tornado_py3.platform.auto import set_close_exec as set_close_exec from tornado_py3.util import Configurable as Configurable, errno_from_exception as errno_from_exception from typing import Any, Awaitable, Callable, Dict, List, Tuple, Type, Union def bind_sockets(port: int, address: str=..., family: socket.AddressFamily=..., backlog: int=..., flags: int=..., reuse_port: bool=...) -> List[socket.socket]: ... def bind_unix_socket(file: str, mode: int=..., backlog: int=...) -> socket.socket: ... def add_accept_handler(sock: socket.socket, callback: Callable[[socket.socket, Any], None]) -> Callable[[], None]: ... def is_valid_ip(ip: str) -> bool: ... class Resolver(Configurable): @classmethod def configurable_base(cls: Any) -> Type[Resolver]: ... @classmethod def configurable_default(cls: Any) -> Type[Resolver]: ... def resolve(self, host: str, port: int, family: socket.AddressFamily=...) -> Awaitable[List[Tuple[int, Any]]]: ... def close(self) -> None: ... class DefaultExecutorResolver(Resolver): async def resolve(self, host: str, port: int, family: socket.AddressFamily=...) -> List[Tuple[int, Any]]: ... class ExecutorResolver(Resolver): io_loop: Any = ... executor: Any = ... close_executor: Any = ... def initialize(self, executor: concurrent.futures.Executor=..., close_executor: bool=...) -> None: ... def close(self) -> None: ... def resolve(self, host: str, port: int, family: socket.AddressFamily=...) -> List[Tuple[int, Any]]: ... class BlockingResolver(ExecutorResolver): def initialize(self) -> None: ... class ThreadedResolver(ExecutorResolver): def initialize(self, num_threads: int=...) -> None: ... class OverrideResolver(Resolver): resolver: Any = ... mapping: Any = ... def initialize(self, resolver: Resolver, mapping: dict) -> None: ... def close(self) -> None: ... def resolve(self, host: str, port: int, family: socket.AddressFamily=...) -> Awaitable[List[Tuple[int, Any]]]: ... def ssl_options_to_context(ssl_options: Union[Dict[str, Any], ssl.SSLContext]) -> ssl.SSLContext: ... def ssl_wrap_socket(socket: socket.socket, ssl_options: Union[Dict[str, Any], ssl.SSLContext], server_hostname: str=..., **kwargs: Any) -> ssl.SSLSocket: ...