Messages Module

The messages module provides the foundation for Chanx's structured message system, including base message types, channel events, 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.BaseChannelEvent(*, handler: Any, payload: Any)

Base class for typed channel events.

Channel events provide a way to send typed messages through the channel layer to specific consumer methods. Each event type must define a unique 'handler' field that corresponds to a method name on the target consumer.

handler

Method name on the consumer that will handle this event

Type:: Any

payload

Event-specific data payload

Type:: Any

Incoming Messages

class chanx.messages.incoming.PingMessage(*, action: Literal['ping'] = 'ping', payload: None = None): Simple ping message to check connection status.

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

Type:: chanx.messages.outgoing.AuthenticationPayload

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']