libervia-backend: libervia/backend/core/xmpp.py comparison

comparison libervia/backend/core/xmpp.py @ 4151:18026ce0819c

core (xmpp): message reception workflow refactoring: - Call methods from a root async one instead of using Deferred callbacks chain. - Use a queue to be sure to process messages in order.

author	Goffi <goffi@goffi.org>
date	Wed, 22 Nov 2023 14:50:35 +0100
parents	bc7d45dedeb0
children	d67eaa684484

comparison

equal deleted inserted replaced

-:26534d959d2d
+:18026ce0819c
 from libervia.backend.core.i18n import _
 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
 log = getLogger(__name__)
 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):
 return self.parent
 if elt.uri == namespace:
 elt.defaultUri = elt.uri = C.NS_CLIENT
 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 <message> xml
 @param client(SatXMPPClient, None): client to map message id to uid
 if None, mapping will not be done
 message_elt['to'] = client.jid.full()
 message = {}
 subject = {}
 extra = {}
-data = {
+data: MessageData = {
 "from": jid.JID(message_elt["from"]),
 "to": jid.JID(message_elt["to"]),
 "uid": message_elt.getAttribute(
 "uid", str(uuid.uuid4())
 ),  # XXX: uid is not a standard attribute but may be added by plugins
 data["delay_sender"] = parsed_delay.sender.full()
 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
+def onMessage(self, message_elt: domish.Element) -> None:
+message_elt._received_timestamp = time.time()
-It is the first callback called after message_received trigger
+self.messages_queue.put(message_elt)
-@param cont(bool): workflow will continue only if this is True
-@param message_elt(domish.Element): message stanza
+async def process_messages(self) -> None:
-may have be modified by triggers
+"""Process message in order
-@param post_treat(defer.Deferred): post parsing treatments
-"""
+Messages are processed in a queue to avoid race conditions and ensure orderly
-if not cont:
+processing.
-return
+"""
-data = self.parse_message(message_elt)
+client = self.parent
-post_treat.addCallback(self.complete_attachments)
+if client is None:
-post_treat.addCallback(self.skip_empty_message)
+log.error("client should not be None!")
-if not client.is_component or client.receiveHistory:
+raise exceptions.InternalError()
-post_treat.addCallback(
+while True:
-lambda ret: defer.ensureDeferred(self.add_to_history(ret))
+message_elt = await self.messages_queue.get()
-)
+try:
-if not client.is_component:
+await self.process_message(client, message_elt)
-post_treat.addCallback(self.bridge_signal, data)
+except Exception:
-post_treat.addErrback(self.cancel_error_trap)
+log.exception(f"Can't process message {message_elt.toXml()}")
-post_treat.callback(data)
+async def process_message(
-def onMessage(self, message_elt):
+self,
+client: SatXMPPEntity,
+message_elt: domish.Element
+) -> None:
 # TODO: handle threads
-message_elt._received_timestamp = time.time()
-client = self.parent
 if not "from" in message_elt.attributes:
 message_elt["from"] = client.jid.host
 log.debug(_("got message from: {from_}").format(from_=message_elt["from"]))
 if self.client.is_component and message_elt.uri == component.NS_COMPONENT_ACCEPT:
 # we use client namespace all the time to simplify parsing
 self.normalize_ns(message_elt, component.NS_COMPONENT_ACCEPT)
 # plugin can add their treatments to this deferred
 post_treat = defer.Deferred()
+if not await self.host.trigger.async_point(
-d = self.host.trigger.async_point(
 "message_received", client, message_elt, post_treat
-)
+):
+return
-d.addCallback(self._on_message_start_workflow, client, message_elt, post_treat)
+try:
+data = self.parse_message(message_elt)
-def complete_attachments(self, data):
+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: 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:
 name = (Path(unquote(urlparse(attachment['url']).path)).name
 or C.FILE_DEFAULT_NAME)
 if ((C.KEY_ATTACHMENTS_MEDIA_TYPE not in attachment
 and "name" in attachment)):
 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):
+async def add_to_history(self, data: MessageData) -> MessageData:
-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):
 if data.pop("history", None) == C.HISTORY_SKIP:
 log.debug("history is skipped as requested")
 data["extra"]["history"] = C.HISTORY_SKIP
 else:
 # we need a message to store
 if self.parent.is_message_printable(data):
 return await self.host.memory.add_to_history(self.parent, data)
 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"]
 except KeyError:
 pass
 )
 else:
 log.debug("Discarding bridge signal for empty message: {data}".format(
 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):
 def __init__(self, host):

Mercurial > libervia-backend

comparison libervia/backend/core/xmpp.py @ 4151:18026ce0819c