slidge.plugins.mattermost.websocket

Module Contents

Classes

EventType	str(object='') -> str
MattermostEvent
Websocket

Functions

handle_event(d, event_handler)

Attributes

log

class slidge.plugins.mattermost.websocket.EventType

Bases: str, enum.Enum

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

Initialize self. See help(type(self)) for accurate signature.

AddedToTeam = 'added_to_team'

AuthenticationChallenge = 'authentication_challenge'

ChannelConverted = 'channel_converted'

ChannelCreated = 'channel_created'

ChannelDeleted = 'channel_deleted'

ChannelMemberUpdated = 'channel_member_updated'

ChannelUpdated = 'channel_updated'

ChannelViewed = 'channel_viewed'

ConfigChanged = 'config_changed'

DeleteTeam = 'delete_team'

DirectAdded = 'direct_added'

EmojiAdded = 'emoji_added'

EphemeralMessage = 'ephemeral_message'

GroupAdded = 'group_added'

Hello = 'hello'

LeaveTeam = 'leave_team'

LicenseChanged = 'license_changed'

MemberroleUpdated = 'memberrole_updated'

NewUser = 'new_user'

PluginDisabled = 'plugin_disabled'

PluginEnabled = 'plugin_enabled'

PluginStatusesChanged = 'plugin_statuses_changed'

PostDeleted = 'post_deleted'

PostEdited = 'post_edited'

PostUnread = 'post_unread'

Posted = 'posted'

PreferenceChanged = 'preference_changed'

PreferencesChanged = 'preferences_changed'

PreferencesDeleted = 'preferences_deleted'

ReactionAdded = 'reaction_added'

ReactionRemoved = 'reaction_removed'

Response = 'response'

RoleUpdated = 'role_updated'

StatusChange = 'status_change'

Typing = 'typing'

UpdateTeam = 'update_team'

UserAdded = 'user_added'

UserRemoved = 'user_removed'

UserRoleUpdated = 'user_role_updated'

UserUpdated = 'user_updated'

DialogOpened = 'dialog_opened'

ThreadUpdated = 'thread_updated'

ThreadFollowChanged = 'thread_follow_changed'

ThreadReadChanged = 'thread_read_changed'

SidebarCategoryUpdated = 'sidebar_category_updated'

Unknown = '__unknown__'

class slidge.plugins.mattermost.websocket.MattermostEvent

type: EventType

data: dict

broadcast: dict

left: dict

__str__()

Return str(self).

class slidge.plugins.mattermost.websocket.Websocket(url, token)

async connect(event_handler)

Connect to the websocket and authenticate it. When the authentication has finished, start the loop listening for messages, sending a ping to the server to keep the connection alive. :param event_handler: Every websocket event will be passed there. Takes one argument. :type event_handler: Function(message) :return:

async _start_loop(websocket, event_handler)

We will listen for websockets events, sending a heartbeats on a timer. If we don’t the webserver would close the idle connection, forcing us to reconnect.

async _do_heartbeats(websocket)

This is a little complicated, but we only need to pong the websocket if we haven’t received a message inside the timeout window. Since messages can be received, while we are waiting we need to check after sleep.

disconnect()

Sets self._alive to False so the loop in self._start_loop will finish.

async _authenticate_websocket(websocket)

Sends an authentication challenge over a websocket. This is not needed when we just send the cookie we got on login when connecting to the websocket.

async user_typing(channel_id)

async get_statuses()

async get_statuses_by_ids(ids)

Parameters:

ids (list[str]) –

async slidge.plugins.mattermost.websocket.handle_event(d, event_handler)

slidge.plugins.mattermost.websocket.log