SickGear/lib/tornado/iostream.pyi

import socket
import ssl
from tornado_py3 import ioloop as ioloop
from tornado_py3.concurrent import Future as Future, future_set_result_unless_cancelled as future_set_result_unless_cancelled
from tornado_py3.log import gen_log as gen_log
from tornado_py3.netutil import ssl_wrap_socket as ssl_wrap_socket
from tornado_py3.util import errno_from_exception as errno_from_exception
from types import TracebackType
from typing import Any, Awaitable, Callable, Dict, Optional, Tuple, Type, Union

class StreamClosedError(IOError):
    real_error: Any = ...
    def __init__(self, real_error: BaseException=...) -> None: ...

class UnsatisfiableReadError(Exception): ...
class StreamBufferFullError(Exception): ...

class _StreamBuffer:
    def __init__(self) -> None: ...
    def __len__(self) -> int: ...
    def append(self, data: Union[bytes, bytearray, memoryview]) -> None: ...
    def peek(self, size: int) -> memoryview: ...
    def advance(self, size: int) -> None: ...

class BaseIOStream:
    io_loop: Any = ...
    max_buffer_size: Any = ...
    read_chunk_size: Any = ...
    max_write_buffer_size: Any = ...
    error: Any = ...
    def __init__(self, max_buffer_size: int=..., read_chunk_size: int=..., max_write_buffer_size: int=...) -> None: ...
    def fileno(self) -> Union[int, ioloop._Selectable]: ...
    def close_fd(self) -> None: ...
    def write_to_fd(self, data: memoryview) -> int: ...
    def read_from_fd(self, buf: Union[bytearray, memoryview]) -> Optional[int]: ...
    def get_fd_error(self) -> Optional[Exception]: ...
    def read_until_regex(self, regex: bytes, max_bytes: int=...) -> Awaitable[bytes]: ...
    def read_until(self, delimiter: bytes, max_bytes: int=...) -> Awaitable[bytes]: ...
    def read_bytes(self, num_bytes: int, partial: bool=...) -> Awaitable[bytes]: ...
    def read_into(self, buf: bytearray, partial: bool=...) -> Awaitable[int]: ...
    def read_until_close(self) -> Awaitable[bytes]: ...
    def write(self, data: Union[bytes, memoryview]) -> Future[None]: ...
    def set_close_callback(self, callback: Optional[Callable[[], None]]) -> None: ...
    def close(self, exc_info: Union[None, bool, BaseException, Tuple[Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]]]=...) -> None: ...
    def reading(self) -> bool: ...
    def writing(self) -> bool: ...
    def closed(self) -> bool: ...
    def set_nodelay(self, value: bool) -> None: ...

class IOStream(BaseIOStream):
    socket: Any = ...
    def __init__(self, socket: socket.socket, *args: Any, **kwargs: Any) -> None: ...
    def fileno(self) -> Union[int, ioloop._Selectable]: ...
    def close_fd(self) -> None: ...
    def get_fd_error(self) -> Optional[Exception]: ...
    def read_from_fd(self, buf: Union[bytearray, memoryview]) -> Optional[int]: ...
    def write_to_fd(self, data: memoryview) -> int: ...
    def connect(self, address: tuple, server_hostname: str=...) -> Future[_IOStreamType]: ...
    def start_tls(self, server_side: bool, ssl_options: Union[Dict[str, Any], ssl.SSLContext]=..., server_hostname: str=...) -> Awaitable[SSLIOStream]: ...
    def set_nodelay(self, value: bool) -> None: ...

class SSLIOStream(IOStream):
    socket: ssl.SSLSocket = ...
    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
    def reading(self) -> bool: ...
    def writing(self) -> bool: ...
    def connect(self, address: Tuple, server_hostname: str=...) -> Future[SSLIOStream]: ...
    def wait_for_handshake(self) -> Future[SSLIOStream]: ...
    def write_to_fd(self, data: memoryview) -> int: ...
    def read_from_fd(self, buf: Union[bytearray, memoryview]) -> Optional[int]: ...

class PipeIOStream(BaseIOStream):
    fd: Any = ...
    def __init__(self, fd: int, *args: Any, **kwargs: Any) -> None: ...
    def fileno(self) -> int: ...
    def close_fd(self) -> None: ...
    def write_to_fd(self, data: memoryview) -> int: ...
    def read_from_fd(self, buf: Union[bytearray, memoryview]) -> Optional[int]: ...

def doctests() -> Any: ...