# Stubs for autobahn.websocket.protocol (Python 3) # # NOTE: This dynamically typed stub was automatically generated by stubgen. from autobahn.util import ObservableMixin from typing import Any, Optional xrange = range class TrafficStats: def __init__(self) -> None: ... outgoingOctetsWireLevel: int = ... outgoingOctetsWebSocketLevel: int = ... outgoingOctetsAppLevel: int = ... outgoingWebSocketFrames: int = ... outgoingWebSocketMessages: int = ... incomingOctetsWireLevel: int = ... incomingOctetsWebSocketLevel: int = ... incomingOctetsAppLevel: int = ... incomingWebSocketFrames: int = ... incomingWebSocketMessages: int = ... preopenOutgoingOctetsWireLevel: int = ... preopenIncomingOctetsWireLevel: int = ... def reset(self) -> None: ... def __json__(self): ... class FrameHeader: opcode: Any = ... fin: Any = ... rsv: Any = ... length: Any = ... mask: Any = ... def __init__(self, opcode: Any, fin: Any, rsv: Any, length: Any, mask: Any) -> None: ... class Timings: def __init__(self) -> None: ... def track(self, key: Any) -> None: ... def diff(self, startKey: Any, endKey: Any, formatted: bool = ...): ... def __getitem__(self, key: Any): ... def __iter__(self): ... class WebSocketProtocol(ObservableMixin): peer: str = ... SUPPORTED_SPEC_VERSIONS: Any = ... SUPPORTED_PROTOCOL_VERSIONS: Any = ... SPEC_TO_PROTOCOL_VERSION: Any = ... PROTOCOL_TO_SPEC_VERSION: Any = ... DEFAULT_SPEC_VERSION: int = ... MESSAGE_TYPE_TEXT: int = ... MESSAGE_TYPE_BINARY: int = ... STATE_CLOSED: int = ... STATE_CONNECTING: int = ... STATE_CLOSING: int = ... STATE_OPEN: int = ... STATE_PROXY_CONNECTING: int = ... SEND_STATE_GROUND: int = ... SEND_STATE_MESSAGE_BEGIN: int = ... SEND_STATE_INSIDE_MESSAGE: int = ... SEND_STATE_INSIDE_MESSAGE_FRAME: int = ... CLOSE_STATUS_CODE_NORMAL: int = ... CLOSE_STATUS_CODE_GOING_AWAY: int = ... CLOSE_STATUS_CODE_PROTOCOL_ERROR: int = ... CLOSE_STATUS_CODE_UNSUPPORTED_DATA: int = ... CLOSE_STATUS_CODE_RESERVED1: int = ... CLOSE_STATUS_CODE_NULL: int = ... CLOSE_STATUS_CODE_ABNORMAL_CLOSE: int = ... CLOSE_STATUS_CODE_INVALID_PAYLOAD: int = ... CLOSE_STATUS_CODE_POLICY_VIOLATION: int = ... CLOSE_STATUS_CODE_MESSAGE_TOO_BIG: int = ... CLOSE_STATUS_CODE_MANDATORY_EXTENSION: int = ... CLOSE_STATUS_CODE_INTERNAL_ERROR: int = ... CLOSE_STATUS_CODE_SERVICE_RESTART: int = ... CLOSE_STATUS_CODE_TRY_AGAIN_LATER: int = ... CLOSE_STATUS_CODE_UNASSIGNED1: int = ... CLOSE_STATUS_CODE_TLS_HANDSHAKE_FAILED: int = ... CLOSE_STATUS_CODES_ALLOWED: Any = ... CONFIG_ATTRS_COMMON: Any = ... CONFIG_ATTRS_SERVER: Any = ... CONFIG_ATTRS_CLIENT: Any = ... is_closed: Any = ... is_open: Any = ... def __init__(self) -> None: ... def onOpen(self) -> None: ... message_is_binary: Any = ... message_data: Any = ... message_data_total_length: int = ... def onMessageBegin(self, isBinary: Any) -> None: ... frame_length: Any = ... frame_data: Any = ... wasMaxMessagePayloadSizeExceeded: bool = ... wasMaxFramePayloadSizeExceeded: bool = ... def onMessageFrameBegin(self, length: Any) -> None: ... def onMessageFrameData(self, payload: Any) -> None: ... def onMessageFrameEnd(self) -> None: ... def onMessageFrame(self, payload: Any) -> None: ... def onMessageEnd(self) -> None: ... def onMessage(self, payload: Any, isBinary: Any) -> None: ... def onPing(self, payload: Any) -> None: ... def onPong(self, payload: Any) -> None: ... def onClose(self, wasClean: Any, code: Any, reason: Any) -> None: ... remoteCloseCode: Any = ... remoteCloseReason: Any = ... closeHandshakeTimeoutCall: Any = ... wasClean: bool = ... serverConnectionDropTimeoutCall: Any = ... def onCloseFrame(self, code: Any, reasonRaw: Any): ... wasNotCleanReason: str = ... wasServerConnectionDropTimeout: bool = ... def onServerConnectionDropTimeout(self) -> None: ... openHandshakeTimeoutCall: Any = ... wasOpenHandshakeTimeout: bool = ... def onOpenHandshakeTimeout(self) -> None: ... wasCloseHandshakeTimeout: bool = ... def onCloseHandshakeTimeout(self) -> None: ... autoPingTimeoutCall: Any = ... def onAutoPingTimeout(self) -> None: ... droppedByMe: bool = ... state: Any = ... def dropConnection(self, abort: bool = ...) -> None: ... trackTimings: Any = ... trackedTimings: Any = ... def setTrackTimings(self, enable: Any) -> None: ... def logRxOctets(self, data: Any) -> None: ... def logTxOctets(self, data: Any, sync: Any) -> None: ... def logRxFrame(self, frameHeader: Any, payload: Any) -> None: ... def logTxFrame(self, frameHeader: Any, payload: Any, repeatLength: Any, chopsize: Any, sync: Any) -> None: ... def consumeData(self) -> None: ... def processProxyConnect(self) -> None: ... def processHandshake(self) -> None: ... def sendData(self, data: Any, sync: bool = ..., chopsize: Optional[Any] = ...) -> None: ... def sendPreparedMessage(self, preparedMsg: Any) -> None: ... current_frame_masker: Any = ... data: Any = ... current_frame: Any = ... def processData(self): ... control_frame_data: Any = ... inside_message: bool = ... utf8validateIncomingCurrentMessage: bool = ... utf8validateLast: Any = ... def onFrameBegin(self) -> None: ... def onFrameData(self, payload: Any): ... def onFrameEnd(self): ... autoPingPending: Any = ... autoPingPendingCall: Any = ... def processControlFrame(self): ... def sendFrame(self, opcode: Any, payload: bytes = ..., fin: bool = ..., rsv: int = ..., mask: Optional[Any] = ..., payload_len: Optional[Any] = ..., chopsize: Optional[Any] = ..., sync: bool = ...) -> None: ... def sendPing(self, payload: Optional[Any] = ...) -> None: ... def sendPong(self, payload: Optional[Any] = ...) -> None: ... closedByMe: Any = ... localCloseCode: Any = ... localCloseReason: Any = ... def sendCloseFrame(self, code: Optional[Any] = ..., reasonUtf8: Optional[Any] = ..., isReply: bool = ...) -> None: ... def sendClose(self, code: Optional[Any] = ..., reason: Optional[Any] = ...) -> None: ... send_message_opcode: Any = ... send_state: Any = ... send_compressed: bool = ... def beginMessage(self, isBinary: bool = ..., doNotCompress: bool = ...) -> None: ... send_message_frame_length: Any = ... send_message_frame_mask: Any = ... send_message_frame_masker: Any = ... def beginMessageFrame(self, length: Any) -> None: ... def sendMessageFrameData(self, payload: Any, sync: bool = ...): ... def endMessage(self) -> None: ... def sendMessageFrame(self, payload: Any, sync: bool = ...) -> None: ... def sendMessage(self, payload: Any, isBinary: bool = ..., fragmentSize: Optional[Any] = ..., sync: bool = ..., doNotCompress: bool = ...) -> None: ... class PreparedMessage: payload: Any = ... binary: Any = ... doNotCompress: Any = ... payloadHybi: Any = ... def __init__(self, payload: Any, isBinary: Any, applyMask: Any, doNotCompress: Any) -> None: ... class WebSocketFactory: def prepareMessage(self, payload: Any, isBinary: bool = ..., doNotCompress: bool = ...): ... class WebSocketServerProtocol(WebSocketProtocol): log: Any = ... CONFIG_ATTRS: Any = ... def onConnect(self, request: Any) -> None: ... def processProxyConnect(self) -> None: ... http_request_data: Any = ... peer: Any = ... http_request_uri: Any = ... http_request_path: Any = ... http_request_params: Any = ... http_request_host: Any = ... websocket_version: Any = ... websocket_protocols: Any = ... websocket_origin: str = ... websocket_extensions: Any = ... data: Any = ... wasServingFlashSocketPolicyFile: bool = ... def processHandshake(self): ... websocket_protocol_in_use: Any = ... websocket_extensions_in_use: Any = ... http_response_data: Any = ... state: Any = ... openHandshakeTimeoutCall: Any = ... inside_message: bool = ... current_frame: Any = ... autoPingPendingCall: Any = ... def succeedHandshake(self, res: Any): ... wasNotCleanReason: Any = ... def failHandshake(self, reason: Any, code: int = ..., responseHeaders: Optional[Any] = ...) -> None: ... def sendHttpErrorResponse(self, code: Any, reason: Any, responseHeaders: Optional[Any] = ...) -> None: ... def sendHtml(self, html: Any) -> None: ... def sendRedirect(self, url: Any) -> None: ... def sendServerStatus(self, redirectUrl: Optional[Any] = ..., redirectAfter: int = ...) -> None: ... class WebSocketServerFactory(WebSocketFactory): log: Any = ... protocol: Any = ... isServer: bool = ... logOctets: bool = ... logFrames: bool = ... trackTimings: bool = ... countConnections: int = ... def __init__(self, url: Optional[Any] = ..., protocols: Optional[Any] = ..., server: Any = ..., headers: Optional[Any] = ..., externalPort: Optional[Any] = ...) -> None: ... url: Any = ... isSecure: Any = ... host: Any = ... port: Any = ... resource: Any = ... path: Any = ... params: Any = ... protocols: Any = ... server: Any = ... headers: Any = ... externalPort: Any = ... def setSessionParameters(self, url: Optional[Any] = ..., protocols: Optional[Any] = ..., server: Optional[Any] = ..., headers: Optional[Any] = ..., externalPort: Optional[Any] = ...) -> None: ... versions: Any = ... webStatus: bool = ... utf8validateIncoming: bool = ... requireMaskedClientFrames: bool = ... maskServerFrames: bool = ... applyMask: bool = ... maxFramePayloadSize: int = ... maxMessagePayloadSize: int = ... autoFragmentSize: int = ... failByDrop: bool = ... echoCloseCodeReason: bool = ... openHandshakeTimeout: int = ... closeHandshakeTimeout: int = ... tcpNoDelay: bool = ... serveFlashSocketPolicy: bool = ... flashSocketPolicy: str = ... perMessageCompressionAccept: Any = ... autoPingInterval: int = ... autoPingTimeout: int = ... autoPingSize: int = ... allowedOrigins: Any = ... allowedOriginsPatterns: Any = ... allowNullOrigin: bool = ... maxConnections: int = ... trustXForwardedFor: int = ... def resetProtocolOptions(self) -> None: ... def setProtocolOptions(self, versions: Optional[Any] = ..., webStatus: Optional[Any] = ..., utf8validateIncoming: Optional[Any] = ..., maskServerFrames: Optional[Any] = ..., requireMaskedClientFrames: Optional[Any] = ..., applyMask: Optional[Any] = ..., maxFramePayloadSize: Optional[Any] = ..., maxMessagePayloadSize: Optional[Any] = ..., autoFragmentSize: Optional[Any] = ..., failByDrop: Optional[Any] = ..., echoCloseCodeReason: Optional[Any] = ..., openHandshakeTimeout: Optional[Any] = ..., closeHandshakeTimeout: Optional[Any] = ..., tcpNoDelay: Optional[Any] = ..., perMessageCompressionAccept: Optional[Any] = ..., autoPingInterval: Optional[Any] = ..., autoPingTimeout: Optional[Any] = ..., autoPingSize: Optional[Any] = ..., serveFlashSocketPolicy: Optional[Any] = ..., flashSocketPolicy: Optional[Any] = ..., allowedOrigins: Optional[Any] = ..., allowNullOrigin: bool = ..., maxConnections: Optional[Any] = ..., trustXForwardedFor: Optional[Any] = ...) -> None: ... def getConnectionCount(self): ... class WebSocketClientProtocol(WebSocketProtocol): log: Any = ... CONFIG_ATTRS: Any = ... def onConnecting(self, transport_details: Any) -> None: ... def onConnect(self, response: Any) -> None: ... def startProxyConnect(self) -> None: ... data: Any = ... state: Any = ... def processProxyConnect(self): ... def failProxyConnect(self, reason: Any) -> None: ... def startHandshake(self): ... http_response_data: Any = ... websocket_extensions_in_use: Any = ... websocket_protocol_in_use: Any = ... openHandshakeTimeoutCall: Any = ... inside_message: bool = ... current_frame: Any = ... websocket_version: Any = ... autoPingPendingCall: Any = ... def processHandshake(self): ... wasNotCleanReason: Any = ... def failHandshake(self, reason: Any) -> None: ... class WebSocketClientFactory(WebSocketFactory): log: Any = ... protocol: Any = ... isServer: bool = ... logOctets: bool = ... logFrames: bool = ... trackTimings: bool = ... def __init__(self, url: Optional[Any] = ..., origin: Optional[Any] = ..., protocols: Optional[Any] = ..., useragent: Any = ..., headers: Optional[Any] = ..., proxy: Optional[Any] = ...) -> None: ... url: Any = ... isSecure: Any = ... host: Any = ... port: Any = ... resource: Any = ... path: Any = ... params: Any = ... origin: Any = ... protocols: Any = ... useragent: Any = ... headers: Any = ... proxy: Any = ... def setSessionParameters(self, url: Optional[Any] = ..., origin: Optional[Any] = ..., protocols: Optional[Any] = ..., useragent: Optional[Any] = ..., headers: Optional[Any] = ..., proxy: Optional[Any] = ...) -> None: ... version: Any = ... utf8validateIncoming: bool = ... acceptMaskedServerFrames: bool = ... maskClientFrames: bool = ... applyMask: bool = ... maxFramePayloadSize: int = ... maxMessagePayloadSize: int = ... autoFragmentSize: int = ... failByDrop: bool = ... echoCloseCodeReason: bool = ... serverConnectionDropTimeout: int = ... openHandshakeTimeout: int = ... closeHandshakeTimeout: int = ... tcpNoDelay: bool = ... perMessageCompressionOffers: Any = ... perMessageCompressionAccept: Any = ... autoPingInterval: int = ... autoPingTimeout: int = ... autoPingSize: int = ... def resetProtocolOptions(self) -> None: ... def setProtocolOptions(self, version: Optional[Any] = ..., utf8validateIncoming: Optional[Any] = ..., acceptMaskedServerFrames: Optional[Any] = ..., maskClientFrames: Optional[Any] = ..., applyMask: Optional[Any] = ..., maxFramePayloadSize: Optional[Any] = ..., maxMessagePayloadSize: Optional[Any] = ..., autoFragmentSize: Optional[Any] = ..., failByDrop: Optional[Any] = ..., echoCloseCodeReason: Optional[Any] = ..., serverConnectionDropTimeout: Optional[Any] = ..., openHandshakeTimeout: Optional[Any] = ..., closeHandshakeTimeout: Optional[Any] = ..., tcpNoDelay: Optional[Any] = ..., perMessageCompressionOffers: Optional[Any] = ..., perMessageCompressionAccept: Optional[Any] = ..., autoPingInterval: Optional[Any] = ..., autoPingTimeout: Optional[Any] = ..., autoPingSize: Optional[Any] = ...) -> None: ...