Messages Module
The messages module provides the foundation for Chanx's structured message system, including base message
types, containers, and standard message implementations.
Base Messages
- class chanx.messages.base.BaseMessage(*, action: Any, payload: Any)
Base websocket message.
All message types should inherit from this class and define a unique 'action' field using a Literal type.
- action
Discriminator field identifying message type
- Type:
Any
- payload
Optional message payload data
- Type:
Any
- class chanx.messages.base.BaseGroupMessage(*, action: Any, payload: Any, is_mine: bool = False, is_current: bool = False)
Base message for group broadcasting.
Extends BaseMessage with properties to indicate message's relationship to the current user and connection.
- is_mine
Whether message was sent by the current user
- Type:
bool
- is_current
Whether message was sent by the current connection
- Type:
bool
- class chanx.messages.base.BaseIncomingMessage(*, message: Any)
Base WebSocket incoming message wrapper.
This class serves as a container for incoming WebSocket messages, allowing for a discriminated union pattern where the 'message' field can contain any message type derived from BaseMessage.
- message
The wrapped message object, using action as discriminator field
- Type:
Any
Incoming Messages
- class chanx.messages.incoming.PingMessage(*, action: Literal['ping'] = 'ping', payload: None = None)
Simple ping message to check connection status.
- class chanx.messages.incoming.IncomingMessage(*, message: PingMessage)
Ready-to-use implementation of BaseIncomingMessage.
Provides a concrete incoming message container with support for PingMessage only. Can be used directly for simple applications that only need ping functionality, or as a starting point for more complex implementations.
- message
The wrapped message object, using action as discriminator
Outgoing Messages
- class chanx.messages.outgoing.PongMessage(*, action: Literal['pong'] = 'pong', payload: None = None)
Simple pong message to verify connection status.
Used as a reply to ping messages to confirm that the connection is alive.
- class chanx.messages.outgoing.ErrorMessage(*, action: Literal['error'] = 'error', payload: Any = None)
Error message for communicating issues to the client.
Contains error details in the payload field.
- payload
Error information (typically includes a 'detail' field)
- Type:
Any
- class chanx.messages.outgoing.AuthenticationPayload(*, status_code: int, status_text: str, data: Any = None)
Payload for authentication messages.
Contains status information about the authentication process.
- status_code
HTTP-like status code (e.g., 200 for success)
- Type:
int
- status_text
Human-readable status description
- Type:
str
- data
Additional authentication data
- Type:
Any
- class chanx.messages.outgoing.AuthenticationMessage(*, action: Literal['authentication'] = 'authentication', payload: AuthenticationPayload)
Authentication result message sent to the client.
Sent after connection authentication to inform client of the result.
- payload
AuthenticationPayload containing status details
- class chanx.messages.outgoing.CompleteMessage(*, action: Literal['complete'] = 'complete', payload: None = None)
Confirmation message indicating processing is complete.
Sent after a request has been fully processed to signal completion.
- class chanx.messages.outgoing.GroupCompleteMessage(*, action: Literal['group_complete'] = 'group_complete', payload: None = None)
Confirmation message indicating group message processing is complete.
Sent after a group message has been fully processed and distributed to all consumers in the group to signal completion of the group broadcast operation. This allows clients to know when all intended recipients have received the message.
- action
Literal string 'group_complete' as the discriminator value
- Type:
Literal['group_complete']