SickGear/lib/tornado/iostream.pyi

# Stubs for tornado_py3.iostream (Python 3)
#
# NOTE: This dynamically typed stub was automatically generated by stubgen.

import socket
import ssl
from tornado_py3 import ioloop
from tornado_py3.concurrent import Future
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: Optional[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: Optional[int]=..., read_chunk_size: Optional[int]=..., max_write_buffer_size: Optional[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: Optional[int]=...) -> Awaitable[bytes]: ...
    def read_until(self, delimiter: bytes, max_bytes: Optional[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: Optional[str]=...) -> Future[_IOStreamType]: ...
    def start_tls(self, server_side: bool, ssl_options: Optional[Union[Dict[str, Any], ssl.SSLContext]]=..., server_hostname: Optional[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: Optional[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: ...