from abc import ABC
from typing import Iterable, Optional, Union
from platypush.message.event import Event
[docs]
class XmppEvent(Event, ABC):
"""
Base class for XMPP events.
"""
[docs]
def __init__(self, *args, client_jabber_id: str, **kwargs):
"""
:param client_jabber_id: The Jabber ID associated to the client connection.
"""
super().__init__(*args, client_jabber_id=client_jabber_id, **kwargs)
[docs]
class XmppUserEvent(XmppEvent, ABC):
"""
Base class for XMPP user events.
"""
[docs]
def __init__(self, *args, user_id: str, jid: Optional[str] = None, **kwargs):
"""
:param user_id: User ID.
:param jid: The full Jabber ID of the user, if the visibility of the
full ID including the client identifier is available.
"""
jid = jid or user_id
super().__init__(*args, user_id=user_id, jid=jid, **kwargs)
[docs]
class XmppRoomEvent(XmppEvent, ABC):
"""
Base class for XMPP room events.
"""
[docs]
def __init__(self, *args, room_id: str, **kwargs):
"""
:param room_id: Room ID.
"""
super().__init__(*args, room_id=room_id, **kwargs)
[docs]
class XmppConversationEvent(XmppEvent, ABC):
"""
Base class for XMPP p2p conversation events.
"""
[docs]
def __init__(self, *args, conversation_id: str, **kwargs):
"""
:param conversation_id: Conversation ID.
"""
super().__init__(*args, conversation_id=conversation_id, **kwargs)
[docs]
class XmppRoomOccupantEvent(XmppRoomEvent, XmppUserEvent, ABC):
"""
Base class for XMPP events about room members.
"""
[docs]
def __init__(self, *args, is_self: bool, **kwargs):
"""
:param is_self: True if the event is about the current user.
"""
super().__init__(*args, is_self=is_self, **kwargs)
[docs]
class XmppConversationMemberEvent(XmppConversationEvent, XmppUserEvent, ABC):
"""
Base class for XMPP events about conversation members.
"""
[docs]
def __init__(self, *args, is_self: bool, **kwargs):
"""
:param is_self: True if the event is about the current user.
"""
super().__init__(*args, is_self=is_self, **kwargs)
[docs]
class XmppNickChangedEvent(XmppUserEvent, ABC):
"""
Base class for XMPP nick changed events.
"""
[docs]
def __init__(
self, *args, old_nick: Optional[str], new_nick: Optional[str], **kwargs
):
"""
:param old_nick: Old nick.
:param new_nick: New nick.
"""
super().__init__(*args, old_nick=old_nick, new_nick=new_nick, **kwargs)
[docs]
class XmppConnectedEvent(XmppEvent):
"""
Event triggered when the registered XMPP client connects to the server.
"""
[docs]
class XmppDisconnectedEvent(XmppEvent):
"""
Event triggered when the registered XMPP client disconnects from the server.
"""
[docs]
def __init__(self, *args, reason: Optional[Union[str, Exception]] = None, **kwargs):
"""
:param reason: The reason of the disconnection.
"""
super().__init__(*args, reason=str(reason) if reason else None, **kwargs)
[docs]
class XmppUserAvailableEvent(XmppUserEvent):
"""
Event triggered when a user the client is subscribed to becomes available.
"""
[docs]
class XmppUserUnavailableEvent(XmppUserEvent):
"""
Event triggered when a user the client is subscribed to becomes unavailable.
"""
[docs]
class XmppRoomUserAvailableEvent(XmppRoomOccupantEvent):
"""
Event triggered when a user in a joined room becomes available.
"""
[docs]
class XmppRoomUserUnavailableEvent(XmppRoomOccupantEvent):
"""
Event triggered when a user in a joined room becomes unavailable.
"""
[docs]
class XmppMessageReceivedEvent(XmppUserEvent):
"""
Event triggered when the registered XMPP client receives a message.
"""
[docs]
def __init__(self, *args, body: str, **kwargs):
"""
:param body: The body of the message.
"""
super().__init__(*args, body=body, **kwargs)
[docs]
class XmppRoomMessageReceivedEvent(XmppMessageReceivedEvent, XmppRoomOccupantEvent):
"""
Event triggered when a message is received on a multi-user conversation
joined by the client.
"""
[docs]
class XmppRoomInviteAcceptedEvent(XmppRoomEvent):
"""
Event triggered when an invite to a room is accepted.
"""
[docs]
class XmppRoomInviteRejectedEvent(XmppRoomEvent):
"""
Event triggered when an invite to a room is rejected.
"""
[docs]
class XmppRoomJoinEvent(XmppRoomOccupantEvent):
"""
Event triggered when a user joins a room.
"""
[docs]
def __init__(self, *args, members: Optional[Iterable[str]] = None, **kwargs):
"""
:param members: List of IDs of the joined members.
"""
super().__init__(*args, members=list(set(members or [])), **kwargs)
[docs]
class XmppRoomLeaveEvent(XmppRoomOccupantEvent):
"""
Event triggered when a user leaves a room.
"""
[docs]
class XmppRoomEnterEvent(XmppRoomOccupantEvent):
"""
Event triggered when a user first enters a room.
"""
[docs]
class XmppRoomExitEvent(XmppRoomOccupantEvent):
"""
Event triggered when a user exits a room.
"""
[docs]
def __init__(self, *args, reason: Optional[str] = None, **kwargs):
"""
:param reason: Exit reason.
"""
super().__init__(*args, reason=reason, **kwargs)
[docs]
class XmppRoomTopicChangedEvent(XmppRoomEvent):
"""
Event triggered when the topic of a room is changed.
"""
[docs]
def __init__(
self,
*args,
topic: Optional[str] = None,
changed_by: Optional[str] = None,
**kwargs
):
"""
:param topic: New room topic.
:param changed_by: Nick of the user who changed the topic.
"""
super().__init__(*args, topic=topic, changed_by=changed_by, **kwargs)
[docs]
class XmppPresenceChangedEvent(XmppUserEvent):
"""
Event triggered when the reported presence of a user in the contacts list
changes.
"""
[docs]
def __init__(self, *args, status: Optional[str], **kwargs):
"""
:param status: New presence status.
"""
super().__init__(*args, status=status, **kwargs)
[docs]
class XmppRoomPresenceChangedEvent(XmppPresenceChangedEvent, XmppRoomEvent):
"""
Event triggered when the reported presence of a user in a room changes.
"""
[docs]
class XmppRoomAffiliationChangedEvent(XmppRoomOccupantEvent):
"""
Event triggered when the affiliation of a user in a room changes.
"""
[docs]
def __init__(
self,
*args,
affiliation: str,
changed_by: Optional[str] = None,
reason: Optional[str] = None,
**kwargs
):
"""
:param affiliation: New affiliation.
:param changed_by: Nick of the user who changed the affiliation.
:param reason: Affiliation change reason.
"""
super().__init__(
*args,
affiliation=affiliation,
changed_by=changed_by,
reason=reason,
**kwargs
)
[docs]
class XmppRoomRoleChangedEvent(XmppRoomOccupantEvent):
"""
Event triggered when the role of a user in a room changes.
"""
[docs]
def __init__(
self,
*args,
role: str,
changed_by: Optional[str] = None,
reason: Optional[str] = None,
**kwargs
):
"""
:param role: New role.
:param changed_by: Nick of the user who changed the role.
:param reason: Role change reason.
"""
super().__init__(
*args, role=role, changed_by=changed_by, reason=reason, **kwargs
)
[docs]
class XmppRoomNickChangedEvent(XmppNickChangedEvent, XmppRoomOccupantEvent):
"""
Event triggered when a user in a room changes their nick.
"""
[docs]
class XmppRoomInviteEvent(XmppRoomEvent, XmppUserEvent):
"""
Event triggered when the client is invited to join a room.
"""
[docs]
def __init__(
self,
*args,
mode: str,
password: Optional[str] = None,
reason: Optional[str] = None,
**kwargs
):
"""
:param user_id: The user who sent the invite.
:param mode: Invite mode, either ``DIRECT`` or ``MEDIATED``.
:param password: The room password.
:param reason: Optional invite reason.
"""
super().__init__(*args, mode=mode, password=password, reason=reason, **kwargs)
[docs]
class XmppConversationAddedEvent(XmppConversationEvent):
"""
Event triggered when a conversation is added to the client's list.
"""
[docs]
def __init__(self, *args, members: Optional[Iterable[str]] = None, **kwargs):
"""
:param members: Jabber IDs of the conversation members.
"""
super().__init__(*args, members=list(set(members or [])), **kwargs)
[docs]
class XmppConversationEnterEvent(XmppConversationEvent):
"""
Event triggered when the user enters a conversation.
"""
[docs]
class XmppConversationExitEvent(XmppConversationEvent):
"""
Event triggered when the user exits a conversation.
"""
[docs]
class XmppConversationNickChangedEvent(
XmppNickChangedEvent, XmppConversationMemberEvent
):
"""
Event triggered when a user in a p2p conversation changes their nick.
"""
[docs]
class XmppConversationJoinEvent(XmppConversationMemberEvent):
"""
Event triggered when a user enters a conversation.
"""
[docs]
class XmppConversationLeaveEvent(XmppConversationMemberEvent):
"""
Event triggered when the user leaves a conversation.
"""
