SickGear/lib/tornado/websocket.pyi

import abc
import tornado_py3.web
from tornado_py3 import gen as gen, httpclient as httpclient, httputil as httputil, simple_httpclient as simple_httpclient
from tornado_py3.concurrent import Future as Future, future_set_result_unless_cancelled as future_set_result_unless_cancelled
from tornado_py3.escape import native_str as native_str, to_unicode as to_unicode, utf8 as utf8
from tornado_py3.ioloop import IOLoop as IOLoop, PeriodicCallback as PeriodicCallback
from tornado_py3.iostream import IOStream as IOStream, StreamClosedError as StreamClosedError
from tornado_py3.log import app_log as app_log, gen_log as gen_log
from tornado_py3.queues import Queue as Queue
from tornado_py3.tcpclient import TCPClient as TCPClient
from types import TracebackType
from typing import Any, Awaitable, Callable, Dict, List, Optional, Type, Union
from typing_extensions import Protocol as Protocol

class _Compressor:
    def compress(self, data: bytes) -> bytes: ...
    def flush(self, mode: int) -> bytes: ...

class _Decompressor:
    unconsumed_tail: bytes = ...
    def decompress(self, data: bytes, max_length: int) -> bytes: ...

class _WebSocketDelegate:
    def on_ws_connection_close(self, close_code: int=..., close_reason: str=...) -> None: ...
    def on_message(self, message: Union[str, bytes]) -> Optional[Awaitable[None]]: ...
    def on_ping(self, data: bytes) -> None: ...
    def on_pong(self, data: bytes) -> None: ...
    def log_exception(self, typ: Optional[Type[BaseException]], value: Optional[BaseException], tb: Optional[TracebackType]) -> None: ...

class WebSocketError(Exception): ...
class WebSocketClosedError(WebSocketError): ...
class _DecompressTooLargeError(Exception): ...

class _WebSocketParams:
    ping_interval: Any = ...
    ping_timeout: Any = ...
    max_message_size: Any = ...
    compression_options: Any = ...
    def __init__(self, ping_interval: float=..., ping_timeout: float=..., max_message_size: int=..., compression_options: Dict[str, Any]=...) -> None: ...

class WebSocketHandler(tornado_py3.web.RequestHandler):
    ws_connection: Any = ...
    close_code: Any = ...
    close_reason: Any = ...
    stream: Any = ...
    def __init__(self, application: tornado_py3.web.Application, request: httputil.HTTPServerRequest, **kwargs: Any) -> None: ...
    open_args: Any = ...
    open_kwargs: Any = ...
    async def get(self, *args: Any, **kwargs: Any) -> None: ...
    @property
    def ping_interval(self) -> Optional[float]: ...
    @property
    def ping_timeout(self) -> Optional[float]: ...
    @property
    def max_message_size(self) -> int: ...
    def write_message(self, message: Union[bytes, str, Dict[str, Any]], binary: bool=...) -> Future[None]: ...
    def select_subprotocol(self, subprotocols: List[str]) -> Optional[str]: ...
    @property
    def selected_subprotocol(self) -> Optional[str]: ...
    def get_compression_options(self) -> Optional[Dict[str, Any]]: ...
    def open(self, *args: str, **kwargs: str) -> Optional[Awaitable[None]]: ...
    def on_message(self, message: Union[str, bytes]) -> Optional[Awaitable[None]]: ...
    def ping(self, data: Union[str, bytes]=...) -> None: ...
    def on_pong(self, data: bytes) -> None: ...
    def on_ping(self, data: bytes) -> None: ...
    def on_close(self) -> None: ...
    def close(self, code: int=..., reason: str=...) -> None: ...
    def check_origin(self, origin: str) -> bool: ...
    def set_nodelay(self, value: bool) -> None: ...
    def on_connection_close(self) -> None: ...
    def on_ws_connection_close(self, close_code: int=..., close_reason: str=...) -> None: ...
    def send_error(self, *args: Any, **kwargs: Any) -> None: ...
    def get_websocket_protocol(self) -> Optional[WebSocketProtocol]: ...

class WebSocketProtocol(abc.ABC, metaclass=abc.ABCMeta):
    handler: Any = ...
    stream: Any = ...
    client_terminated: bool = ...
    server_terminated: bool = ...
    def __init__(self, handler: _WebSocketDelegate) -> None: ...
    def on_connection_close(self) -> None: ...
    @abc.abstractmethod
    def close(self, code: int=..., reason: str=...) -> None: ...
    @abc.abstractmethod
    def is_closing(self) -> bool: ...
    @abc.abstractmethod
    async def accept_connection(self, handler: WebSocketHandler) -> None: ...
    @abc.abstractmethod
    def write_message(self, message: Union[str, bytes], binary: bool=...) -> Future[None]: ...
    @property
    @abc.abstractmethod
    def selected_subprotocol(self) -> Optional[str]: ...
    @abc.abstractmethod
    def write_ping(self, data: bytes) -> None: ...
    @abc.abstractmethod
    def start_pinging(self) -> None: ...
    @abc.abstractmethod
    def set_nodelay(self, x: bool) -> None: ...

class _PerMessageDeflateCompressor:
    def __init__(self, persistent: bool, max_wbits: Optional[int], compression_options: Dict[str, Any]=...) -> None: ...
    def compress(self, data: bytes) -> bytes: ...

class _PerMessageDeflateDecompressor:
    def __init__(self, persistent: bool, max_wbits: Optional[int], max_message_size: int, compression_options: Dict[str, Any]=...) -> None: ...
    def decompress(self, data: bytes) -> bytes: ...

class WebSocketProtocol13(WebSocketProtocol):
    FIN: int = ...
    RSV1: int = ...
    RSV2: int = ...
    RSV3: int = ...
    RSV_MASK: Any = ...
    OPCODE_MASK: int = ...
    stream: IOStream = ...
    mask_outgoing: Any = ...
    params: Any = ...
    ping_callback: Any = ...
    last_ping: float = ...
    last_pong: float = ...
    close_code: Any = ...
    close_reason: Any = ...
    def __init__(self, handler: _WebSocketDelegate, mask_outgoing: bool, params: _WebSocketParams) -> None: ...
    @property
    def selected_subprotocol(self) -> Optional[str]: ...
    @selected_subprotocol.setter
    def selected_subprotocol(self, value: Optional[str]) -> None: ...
    async def accept_connection(self, handler: WebSocketHandler) -> None: ...
    @staticmethod
    def compute_accept_value(key: Union[str, bytes]) -> str: ...
    def write_message(self, message: Union[str, bytes], binary: bool=...) -> Future[None]: ...
    def write_ping(self, data: bytes) -> None: ...
    server_terminated: bool = ...
    def close(self, code: int=..., reason: str=...) -> None: ...
    def is_closing(self) -> bool: ...
    @property
    def ping_interval(self) -> Optional[float]: ...
    @property
    def ping_timeout(self) -> Optional[float]: ...
    def start_pinging(self) -> None: ...
    def periodic_ping(self) -> None: ...
    def set_nodelay(self, x: bool) -> None: ...

class WebSocketClientConnection(simple_httpclient._HTTPConnection):
    protocol: WebSocketProtocol = ...
    connect_future: Any = ...
    read_queue: Any = ...
    key: Any = ...
    close_code: Any = ...
    close_reason: Any = ...
    params: Any = ...
    tcp_client: Any = ...
    def __init__(self, request: httpclient.HTTPRequest, on_message_callback: Callable[[Union[None, str, bytes]], None]=..., compression_options: Dict[str, Any]=..., ping_interval: float=..., ping_timeout: float=..., max_message_size: int=..., subprotocols: Optional[List[str]]=...) -> None: ...
    def close(self, code: int=..., reason: str=...) -> None: ...
    def on_connection_close(self) -> None: ...
    def on_ws_connection_close(self, close_code: int=..., close_reason: str=...) -> None: ...
    headers: Any = ...
    final_callback: Any = ...
    async def headers_received(self, start_line: Union[httputil.RequestStartLine, httputil.ResponseStartLine], headers: httputil.HTTPHeaders) -> None: ...
    def write_message(self, message: Union[str, bytes], binary: bool=...) -> Future[None]: ...
    def read_message(self, callback: Callable[[Future[Union[None, str, bytes]]], None]=...) -> Awaitable[Union[None, str, bytes]]: ...
    def on_message(self, message: Union[str, bytes]) -> Optional[Awaitable[None]]: ...
    def ping(self, data: bytes=...) -> None: ...
    def on_pong(self, data: bytes) -> None: ...
    def on_ping(self, data: bytes) -> None: ...
    def get_websocket_protocol(self) -> WebSocketProtocol: ...
    @property
    def selected_subprotocol(self) -> Optional[str]: ...
    def log_exception(self, typ: Optional[Type[BaseException]], value: Optional[BaseException], tb: Optional[TracebackType]) -> None: ...

def websocket_connect(url: Union[str, httpclient.HTTPRequest], callback: Callable[[Future[WebSocketClientConnection]], None]=..., connect_timeout: float=..., on_message_callback: Callable[[Union[None, str, bytes]], None]=..., compression_options: Dict[str, Any]=..., ping_interval: float=..., ping_timeout: float=..., max_message_size: int=..., subprotocols: List[str]=...) -> Awaitable[WebSocketClientConnection]: ...