import logging
import types
from tornado_py3 import gen as gen, httputil as httputil, iostream as iostream
from tornado_py3.concurrent import Future as Future, future_add_done_callback as future_add_done_callback, future_set_result_unless_cancelled as future_set_result_unless_cancelled
from tornado_py3.escape import native_str as native_str, utf8 as utf8
from tornado_py3.log import app_log as app_log, gen_log as gen_log
from tornado_py3.util import GzipDecompressor as GzipDecompressor
from typing import Any, Awaitable, Callable, Optional, Type, Union

class _QuietException(Exception):
    def __init__(self) -> None: ...

class _ExceptionLoggingContext:
    logger: Any = ...
    def __init__(self, logger: logging.Logger) -> None: ...
    def __enter__(self) -> None: ...
    def __exit__(self, typ: Optional[Type[BaseException]], value: Optional[BaseException], tb: types.TracebackType) -> None: ...

class HTTP1ConnectionParameters:
    no_keep_alive: Any = ...
    chunk_size: Any = ...
    max_header_size: Any = ...
    header_timeout: Any = ...
    max_body_size: Any = ...
    body_timeout: Any = ...
    decompress: Any = ...
    def __init__(self, no_keep_alive: bool=..., chunk_size: int=..., max_header_size: int=..., header_timeout: float=..., max_body_size: int=..., body_timeout: float=..., decompress: bool=...) -> None: ...

class HTTP1Connection(httputil.HTTPConnection):
    is_client: Any = ...
    stream: Any = ...
    params: Any = ...
    context: Any = ...
    no_keep_alive: Any = ...
    def __init__(self, stream: iostream.IOStream, is_client: bool, params: HTTP1ConnectionParameters=..., context: object=...) -> None: ...
    def read_response(self, delegate: httputil.HTTPMessageDelegate) -> Awaitable[bool]: ...
    def set_close_callback(self, callback: Optional[Callable[[], None]]) -> None: ...
    def close(self) -> None: ...
    def detach(self) -> iostream.IOStream: ...
    def set_body_timeout(self, timeout: float) -> None: ...
    def set_max_body_size(self, max_body_size: int) -> None: ...
    def write_headers(self, start_line: Union[httputil.RequestStartLine, httputil.ResponseStartLine], headers: httputil.HTTPHeaders, chunk: bytes=...) -> Future[None]: ...
    def write(self, chunk: bytes) -> Future[None]: ...
    def finish(self) -> None: ...

class _GzipMessageDelegate(httputil.HTTPMessageDelegate):
    def __init__(self, delegate: httputil.HTTPMessageDelegate, chunk_size: int) -> None: ...
    def headers_received(self, start_line: Union[httputil.RequestStartLine, httputil.ResponseStartLine], headers: httputil.HTTPHeaders) -> Optional[Awaitable[None]]: ...
    async def data_received(self, chunk: bytes) -> None: ...
    def finish(self) -> None: ...
    def on_connection_close(self) -> None: ...

class HTTP1ServerConnection:
    stream: Any = ...
    params: Any = ...
    context: Any = ...
    def __init__(self, stream: iostream.IOStream, params: HTTP1ConnectionParameters=..., context: object=...) -> None: ...
    async def close(self) -> None: ...
    def start_serving(self, delegate: httputil.HTTPServerConnectionDelegate) -> None: ...