diff --git a/libervia/backend/core/xmpp.py b/libervia/backend/core/xmpp.py index 9d8068f8..cb7afcbb 100644 --- a/libervia/backend/core/xmpp.py +++ b/libervia/backend/core/xmpp.py @@ -49,6 +49,7 @@ from libervia.backend.core.log import getLogger from libervia.backend.memory import cache from libervia.backend.memory import encryption from libervia.backend.memory import persistent +from libervia.backend.models.core import MessageData from libervia.backend.tools import xml_tools from libervia.backend.tools import utils from libervia.backend.tools.common import data_format @@ -1224,6 +1225,11 @@ class SatMessageProtocol(xmppim.MessageProtocol): def __init__(self, host): xmppim.MessageProtocol.__init__(self) self.host = host + self.messages_queue = defer.DeferredQueue() + + def setHandlerParent(self, parent): + super().setHandlerParent(parent) + defer.ensureDeferred(self.process_messages()) @property def client(self): @@ -1235,7 +1241,7 @@ class SatMessageProtocol(xmppim.MessageProtocol): for child in elt.elements(): self.normalize_ns(child, namespace) - def parse_message(self, message_elt): + def parse_message(self, message_elt: domish.Element) -> MessageData: """Parse a message XML and return message_data @param message_elt(domish.Element): raw xml @@ -1265,7 +1271,7 @@ class SatMessageProtocol(xmppim.MessageProtocol): message = {} subject = {} extra = {} - data = { + data: MessageData = { "from": jid.JID(message_elt["from"]), "to": jid.JID(message_elt["to"]), "uid": message_elt.getAttribute( @@ -1316,33 +1322,34 @@ class SatMessageProtocol(xmppim.MessageProtocol): self.host.trigger.point("message_parse", client, message_elt, data) return data - def _on_message_start_workflow(self, cont, client, message_elt, post_treat): - """Parse message and do post treatments - It is the first callback called after message_received trigger - @param cont(bool): workflow will continue only if this is True - @param message_elt(domish.Element): message stanza - may have be modified by triggers - @param post_treat(defer.Deferred): post parsing treatments - """ - if not cont: - return - data = self.parse_message(message_elt) - post_treat.addCallback(self.complete_attachments) - post_treat.addCallback(self.skip_empty_message) - if not client.is_component or client.receiveHistory: - post_treat.addCallback( - lambda ret: defer.ensureDeferred(self.add_to_history(ret)) - ) - if not client.is_component: - post_treat.addCallback(self.bridge_signal, data) - post_treat.addErrback(self.cancel_error_trap) - post_treat.callback(data) - - def onMessage(self, message_elt): - # TODO: handle threads + def onMessage(self, message_elt: domish.Element) -> None: message_elt._received_timestamp = time.time() + self.messages_queue.put(message_elt) + + async def process_messages(self) -> None: + """Process message in order + + Messages are processed in a queue to avoid race conditions and ensure orderly + processing. + """ client = self.parent + if client is None: + log.error("client should not be None!") + raise exceptions.InternalError() + while True: + message_elt = await self.messages_queue.get() + try: + await self.process_message(client, message_elt) + except Exception: + log.exception(f"Can't process message {message_elt.toXml()}") + + async def process_message( + self, + client: SatXMPPEntity, + message_elt: domish.Element + ) -> None: + # TODO: handle threads if not "from" in message_elt.attributes: message_elt["from"] = client.jid.host log.debug(_("got message from: {from_}").format(from_=message_elt["from"])) @@ -1352,14 +1359,24 @@ class SatMessageProtocol(xmppim.MessageProtocol): # plugin can add their treatments to this deferred post_treat = defer.Deferred() - - d = self.host.trigger.async_point( + if not await self.host.trigger.async_point( "message_received", client, message_elt, post_treat - ) + ): + return + try: + data = self.parse_message(message_elt) - d.addCallback(self._on_message_start_workflow, client, message_elt, post_treat) + self.complete_attachments(data) + if not data["message"] and not data["extra"] and not data["subject"]: + raise exceptions.CancelError("Cancelled empty message") + if not client.is_component or client.receiveHistory: + await self.add_to_history(data) + if not client.is_component: + self.bridge_signal(data) + except exceptions.CancelError: + pass - def complete_attachments(self, data): + def complete_attachments(self, data: MessageData) -> MessageData: """Complete missing metadata of attachments""" for attachment in data['extra'].get(C.KEY_ATTACHMENTS, []): if "name" not in attachment and "url" in attachment: @@ -1371,15 +1388,9 @@ class SatMessageProtocol(xmppim.MessageProtocol): media_type = mimetypes.guess_type(attachment['name'], strict=False)[0] if media_type: attachment[C.KEY_ATTACHMENTS_MEDIA_TYPE] = media_type - return data - def skip_empty_message(self, data): - if not data["message"] and not data["extra"] and not data["subject"]: - raise failure.Failure(exceptions.CancelError("Cancelled empty message")) - return data - - async def add_to_history(self, data): + async def add_to_history(self, data: MessageData) -> MessageData: if data.pop("history", None) == C.HISTORY_SKIP: log.debug("history is skipped as requested") data["extra"]["history"] = C.HISTORY_SKIP @@ -1390,8 +1401,10 @@ class SatMessageProtocol(xmppim.MessageProtocol): else: log.debug("not storing empty message to history: {data}" .format(data=data)) + return data - def bridge_signal(self, __, data): + def bridge_signal(self, data: MessageData) -> MessageData: + """Send signal to frontends for the given message""" try: data["extra"]["received_timestamp"] = str(data["received_timestamp"]) data["extra"]["delay_sender"] = data["delay_sender"] @@ -1417,10 +1430,6 @@ class SatMessageProtocol(xmppim.MessageProtocol): data=data)) return data - def cancel_error_trap(self, failure_): - """A message sending can be cancelled by a plugin treatment""" - failure_.trap(exceptions.CancelError) - class LiberviaRosterProtocol(xmppim.RosterClientProtocol):