Pyjo.WebSocket - The WebSocket protocol

from Pyjo.WebSocket import WS_TEXT, build_frame, parse_frame

data = build_frame(0, 1, 0, 0, 0, WS_TEXT, b'Hello World!')
frame = parse_frame(bytearray(data), 262144)

Pyjo.WebSocket implements the WebSocket protocol as described in RFC 6455.

A frame is a tuple of the FIN bit, the RSV1, RSV2 and RSV3 bits, the opcode and the payload, (fin, rsv1, rsv2, rsv3, op, payload).

Functions

Pyjo.WebSocket implements the following functions, which can be imported individually.

Constants

Pyjo.WebSocket implements the following constants, which can be imported individually.

WS_BINARY

Opcode for Binary frames.

WS_CLOSE

Opcode for Close frames.

WS_CONTINUATION

Opcode for Continuation frames.

WS_PING

Opcode for Ping frames.

WS_PONG

Opcode for Pong frames.

WS_TEXT

Opcode for Text frames.

Debugging

You can set the PYJO_WEBSOCKET_DEBUG environment variable to get some advanced diagnostics information printed to STDERR.

PYJO_WEBSOCKET_DEBUG=1

See also

Pyjoyment, https://mojolicious.org.

Ported from Mojo::WebSocket of Mojolicious 9.49.

Pyjo.WebSocket.build_frame(masked: object, fin: object, rsv1: object, rsv2: object, rsv3: object, op: int, payload: bytes) → bytes

data = build_frame(masked, fin, rsv1, rsv2, rsv3, op, payload)

Build WebSocket frame.

# Masked binary frame with FIN bit and payload
print(build_frame(1, 1, 0, 0, 0, WS_BINARY, b'Hello World!'))

# Text frame with payload but without FIN bit
print(build_frame(0, 0, 0, 0, 0, WS_TEXT, b'Hello '))

# Continuation frame with FIN bit and payload
print(build_frame(0, 1, 0, 0, 0, WS_CONTINUATION, b'World!'))

# Close frame with FIN bit and without payload
print(build_frame(0, 1, 0, 0, 0, WS_CLOSE, b''))

# Ping frame with FIN bit and payload
print(build_frame(0, 1, 0, 0, 0, WS_PING, b'Test 123'))

# Pong frame with FIN bit and payload
print(build_frame(0, 1, 0, 0, 0, WS_PONG, b'Test 123'))

Pyjo.WebSocket.challenge(tx: Pyjo.Transaction.WebSocket.object) → bool

boolean = challenge(Pyjo.Transaction.WebSocket.new())

Check WebSocket handshake challenge.

Pyjo.WebSocket.client_handshake(tx: T) → T

tx = client_handshake(Pyjo.Transaction.HTTP.new())

Perform WebSocket handshake client-side.

Pyjo.WebSocket.parse_frame(buffer: bytearray, max: int) → Frame | bool | None

frame = parse_frame(buffer, limit)

Parse WebSocket frame from a bytearray, which loses the parsed frame. Returns None if no complete frame was found, and True if the frame is bigger than the limit.

# Parse single frame and remove it from buffer
frame = parse_frame(buffer, 262144)
print(f'FIN: {frame[0]}')
print(f'RSV1: {frame[1]}')
print(f'RSV2: {frame[2]}')
print(f'RSV3: {frame[3]}')
print(f'Opcode: {frame[4]}')
print(f'Payload: {frame[5]}')

Pyjo.WebSocket.server_handshake(tx: T) → T

tx = server_handshake(Pyjo.Transaction.HTTP.new())

Perform WebSocket handshake server-side.