twisted.web._websocket_impl

module documentation
(source)

Private implementation module for public twisted.web.websocket.

Class Buffer Undocumented
Class SomeMessage The properties we rely on from the wsproto text and bytes message types.
Function maybeDeliver If it is complete, deliver the given event to the given ...Received() method.
Function serverBootstrap The server requires no bootstrapping, so this is a no-op.
Type Variable CoStr Undocumented
Type Variable SomeStr Undocumented
Variable AnyWSWP Undocumented
Class _WebSocketWireProtocol Undocumented
Function _handle_acceptConnection Undocumented
Function _handle_bytesMessage Undocumented
Function _handle_closeConnection Undocumented
Function _handle_ping Undocumented
Function _handle_pong Undocumented
Function _handle_rejectConnection Undocumented
Function _handle_rejectData Undocumented
Function _handle_textMessage Undocumented
Function _handleEvent Handle a websocket protocol event.
Function _negotiationError Undocumented
Type Variable _WSP Undocumented
Type Alias _Bootstrap Undocumented
Variable _log Undocumented
def maybeDeliver(event: SomeMessage[SomeStr], buffer: Buffer[SomeStr], receivedMethod: Callable[[SomeStr], None]): (source)

If it is complete, deliver the given event to the given ...Received() method.

This re-assembles messages from buffered data because, as wsproto explains, "[a websocket event] only represents a single chunk of data and not a full WebSocket message. You need to buffer and reassemble these chunks to get the full message."

def serverBootstrap(wsc: WSConnection | Connection, t: ITransport): (source)

The server requires no bootstrapping, so this is a no-op.

CoStr = (source)

Undocumented

Value
TypeVar('CoStr', (bytes | bytearray), str,
        covariant=True)
SomeStr = (source)

Undocumented

Value
TypeVar('SomeStr', (bytes | bytearray), str)
AnyWSWP = (source)

Undocumented

@_handleEvent.register
def _handle_acceptConnection(event: AcceptConnection, proto: AnyWSWP): (source)

Undocumented

@_handleEvent.register
def _handle_bytesMessage(event: BytesMessage, proto: AnyWSWP): (source)

Undocumented

@_handleEvent.register
def _handle_closeConnection(event: CloseConnection, proto: AnyWSWP): (source)

Undocumented

@_handleEvent.register
def _handle_ping(event: Ping, proto: AnyWSWP): (source)

Undocumented

@_handleEvent.register
def _handle_pong(event: Pong, proto: AnyWSWP): (source)

Undocumented

@_handleEvent.register
def _handle_rejectConnection(event: RejectConnection, proto: AnyWSWP): (source)

Undocumented

@_handleEvent.register
def _handle_rejectData(event: RejectData, proto: AnyWSWP): (source)

Undocumented

@_handleEvent.register
def _handle_textMessage(event: TextMessage, proto: AnyWSWP): (source)

Undocumented

@singledispatch
def _handleEvent(event: Event, proto: AnyWSWP): (source)

Handle a websocket protocol event.

def _negotiationError(request: Request) -> bytes: (source)

Undocumented

_WSP = (source)

Undocumented

Value
TypeVar('_WSP',
        covariant=True, bound=WebSocketProtocol)
_Bootstrap = (source)

Undocumented

Value
Callable[[(WSConnection | Connection), ITransport], None]
_log = (source)

Undocumented