"""
This type stub file was generated by pyright.
"""

import attr
from typing import Any, Iterable, Optional, Tuple
from multidict import CIMultiDict
from .abc import AbstractStreamWriter
from .http import WSMessage, WSMsgType as WSMsgType, WebSocketWriter
from .typedefs import JSONDecoder, JSONEncoder
from .web_request import BaseRequest
from .web_response import StreamResponse

__all__ = ('WebSocketResponse', 'WebSocketReady', 'WSMsgType')
THRESHOLD_CONNLOST_ACCESS = 5
@attr.s(frozen=True, slots=True)
class WebSocketReady:
    ok = ...
    protocol = ...
    def __bool__(self) -> bool:
        ...
    


class WebSocketResponse(StreamResponse):
    _length_check = ...
    def __init__(self, *, timeout: float = ..., receive_timeout: Optional[float] = ..., autoclose: bool = ..., autoping: bool = ..., heartbeat: Optional[float] = ..., protocols: Iterable[str] = ..., compress: bool = ..., max_msg_size: int = ...) -> None:
        ...
    
    def _cancel_heartbeat(self) -> None:
        ...
    
    def _reset_heartbeat(self) -> None:
        ...
    
    def _send_heartbeat(self) -> None:
        ...
    
    def _pong_not_received(self) -> None:
        ...
    
    async def prepare(self, request: BaseRequest) -> AbstractStreamWriter:
        ...
    
    def _handshake(self, request: BaseRequest) -> Tuple[CIMultiDict[str], str, bool, bool]:
        ...
    
    def _pre_start(self, request: BaseRequest) -> Tuple[str, WebSocketWriter]:
        ...
    
    def _post_start(self, request: BaseRequest, protocol: str, writer: WebSocketWriter) -> None:
        ...
    
    def can_prepare(self, request: BaseRequest) -> WebSocketReady:
        ...
    
    @property
    def closed(self) -> bool:
        ...
    
    @property
    def close_code(self) -> Optional[int]:
        ...
    
    @property
    def ws_protocol(self) -> Optional[str]:
        ...
    
    @property
    def compress(self) -> bool:
        ...
    
    def exception(self) -> Optional[BaseException]:
        ...
    
    async def ping(self, message: bytes = ...) -> None:
        ...
    
    async def pong(self, message: bytes = ...) -> None:
        ...
    
    async def send_str(self, data: str, compress: Optional[bool] = ...) -> None:
        ...
    
    async def send_bytes(self, data: bytes, compress: Optional[bool] = ...) -> None:
        ...
    
    async def send_json(self, data: Any, compress: Optional[bool] = ..., *, dumps: JSONEncoder = ...) -> None:
        ...
    
    async def write_eof(self) -> None:
        ...
    
    async def close(self, *, code: int = ..., message: bytes = ...) -> bool:
        ...
    
    async def receive(self, timeout: Optional[float] = ...) -> WSMessage:
        ...
    
    async def receive_str(self, *, timeout: Optional[float] = ...) -> str:
        ...
    
    async def receive_bytes(self, *, timeout: Optional[float] = ...) -> bytes:
        ...
    
    async def receive_json(self, *, loads: JSONDecoder = ..., timeout: Optional[float] = ...) -> Any:
        ...
    
    async def write(self, data: bytes) -> None:
        ...
    
    def __aiter__(self) -> WebSocketResponse:
        ...
    
    async def __anext__(self) -> WSMessage:
        ...
    


