erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin XEP-0045: dot not wait for MAM retrieval to be completed:

Browse source 
in `_join_MAM`, `room.fully_joined` is called before retrieving the MAM archive, as the
process can be very long, and is not necessary to have the room working (message can be
received after being in the room, and added out of order). This avoid blocking the `join`
workflow for an extended time.

Some renaming and coroutine integrations.
This commit is contained in:
Goffi 2023-03-10 17:22:41 +01:00
commit 1008801fe7
4 changed files with 123 additions and 85 deletions
Download patch file Download diff file Expand all files Collapse all files

172
sat/plugins/plugin_xep_0045.py
View file

@ -17,29 +17,26 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from sat.core.i18n import _, D_
from sat.core.constants import Const as C
from sat.core.log import getLogger
from twisted.internet import defer
from twisted.words.protocols.jabber import jid
from twisted.words.protocols.jabber import error as xmpp_error
from twisted.python import failure
from sat.core import exceptions
from sat.core.xmpp import SatXMPPClient
from sat.memory import memory
import time
from typing import Optional
import uuid
from wokkel import muc, disco, iwokkel
from sat.tools import xml_tools
from zope.interface import implementer
# XXX: mam and rsm come from sat_tmp.wokkel
from twisted.internet import defer
from twisted.python import failure
from twisted.words.protocols.jabber import jid
from twisted.words.protocols.jabber import error as xmpp_error
from wokkel import disco, iwokkel, muc
from wokkel import rsm
from wokkel import mam
from zope.interface import implementer
from sat.core import exceptions
from sat.core.core_types import SatXMPPEntity
from sat.core.constants import Const as C
from sat.core.i18n import D_, _
from sat.core.log import getLogger
from sat.memory import memory
from sat.tools import xml_tools, utils
log = getLogger(__name__)
@ -161,10 +158,8 @@ class XEP_0045(object):
host.trigger.add("messageReceived", self.messageReceivedTrigger, priority=1000000)
host.trigger.add("message_parse", self._message_parseTrigger)
def profileConnected(self, client):
def assign_service(service):
client.muc_service = service
return self.getMUCService(client).addCallback(assign_service)
async def profileConnected(self, client):
client.muc_service = await self.get_MUC_service(client)
def _message_parseTrigger(self, client, message_elt, data):
"""Add stanza-id from the room if present"""
@ -205,7 +200,7 @@ class XEP_0045(object):
return False
return True
def getRoom(self, client: SatXMPPClient, room_jid: jid.JID) -> muc.Room:
def getRoom(self, client: SatXMPPEntity, room_jid: jid.JID) -> muc.Room:
"""Retrieve Room instance from its jid
@param room_jid: jid of the room
@ -224,7 +219,7 @@ class XEP_0045(object):
if room_jid not in client._muc_client.joined_rooms:
raise exceptions.NotFound(_("This room has not been joined"))
def isJoinedRoom(self, client: SatXMPPClient, room_jid: jid.JID) -> bool:
def isJoinedRoom(self, client: SatXMPPEntity, room_jid: jid.JID) -> bool:
"""Tell if a jid is a known and joined room
@room_jid: jid of the room
@ -458,16 +453,18 @@ class XEP_0045(object):
def _getMUCService(self, jid_=None, profile=C.PROF_KEY_NONE):
client = self.host.getClient(profile)
d = self.getMUCService(client, jid_ or None)
d = defer.ensureDeferred(self.get_MUC_service(client, jid_ or None))
d.addCallback(lambda service_jid: service_jid.full() if service_jid is not None else '')
return d
@defer.inlineCallbacks
def getMUCService(self, client, jid_=None):
async def get_MUC_service(
self,
client: SatXMPPEntity,
jid_: Optional[jid.JID] = None) -> Optional[jid.JID]:
"""Return first found MUC service of an entity
@param jid_: entity which may have a MUC service, or None for our own server
@return (jid.JID, None): found service jid or None
@return: found service jid or None
"""
if jid_ is None:
try:
@ -476,8 +473,8 @@ class XEP_0045(object):
pass
else:
# we have a cached value, we return it
defer.returnValue(muc_service)
services = yield self.host.findServiceEntities(client, "conference", "text", jid_)
return muc_service
services = await self.host.findServiceEntities(client, "conference", "text", jid_)
for service in services:
if ".irc." not in service.userhost():
# FIXME:
@ -487,7 +484,7 @@ class XEP_0045(object):
break
else:
muc_service = None
defer.returnValue(muc_service)
return muc_service
def _getUniqueName(self, muc_service="", profile_key=C.PROF_KEY_NONE):
client = self.host.getClient(profile_key)
@ -549,7 +546,13 @@ class XEP_0045(object):
d.addErrback(self._join_eb, client)
return d
def join(self, client, room_jid, nick=None, options=None):
async def join(
self,
client: SatXMPPEntity,
room_jid: jid.JID,
nick: Optional[str] = None,
options: Optional[dict] = None
) -> Optional[muc.Room]:
if not nick:
nick = client.jid.user
if options is None:
@ -558,18 +561,24 @@ class XEP_0045(object):
room = client._muc_client.joined_rooms[room_jid]
log.info(_('{profile} is already in room {room_jid}').format(
profile=client.profile, room_jid = room_jid.userhost()))
return defer.fail(AlreadyJoined(room))
raise AlreadyJoined(room)
log.info(_("[{profile}] is joining room {room} with nick {nick}").format(
profile=client.profile, room=room_jid.userhost(), nick=nick))
self.host.bridge.mucRoomPrepareJoin(room_jid.userhost(), client.profile)
password = options.get("password")
d = client._muc_client.join(room_jid, nick, password)
d.addCallbacks(self._joinCb, self._joinEb,
(client, room_jid, nick),
errbackArgs=(client, room_jid, nick, password))
return d
try:
room = await client._muc_client.join(room_jid, nick, password)
except Exception as e:
room = await utils.asDeferred(
self._joinEb(failure.Failure(e), client, room_jid, nick, password)
)
else:
await defer.ensureDeferred(
self._joinCb(room, client, room_jid, nick)
)
return room
def popRooms(self, client):
"""Remove rooms and return data needed to re-join them
@ -611,7 +620,7 @@ class XEP_0045(object):
def getHandler(self, client):
# create a MUC client and associate it with profile' session
muc_client = client._muc_client = SatMUCClient(self)
muc_client = client._muc_client = LiberviaMUCClient(self)
return muc_client
def kick(self, client, nick, room_jid, options=None):
@ -915,7 +924,7 @@ class XEP_0045(object):
@implementer(iwokkel.IDisco)
class SatMUCClient(muc.MUCClient):
class LiberviaMUCClient(muc.MUCClient):
def __init__(self, plugin_parent):
self.plugin_parent = plugin_parent
@ -967,7 +976,7 @@ class SatMUCClient(muc.MUCClient):
room.state = new_state
def _addRoom(self, room):
super(SatMUCClient, self)._addRoom(room)
super(LiberviaMUCClient, self)._addRoom(room)
room._roster_ok = False # True when occupants list has been fully received
room.state = ROOM_STATE_OCCUPANTS
# FIXME: check if history_d is not redundant with fully_joined
@ -978,14 +987,21 @@ class SatMUCClient(muc.MUCClient):
# we only need to keep last presence status for each jid, so a dict is suitable
room._cache_presence = {}
@defer.inlineCallbacks
def _joinLegacy(self, client, room_jid, nick, password):
async def _join_legacy(
self,
client: SatXMPPEntity,
room_jid: jid.JID,
nick: str,
password: Optional[str]
) -> muc.Room:
"""Join room an retrieve history with legacy method"""
mess_data_list = yield self.host.memory.historyGet(room_jid,
client.jid.userhostJID(),
limit=1,
between=True,
profile=client.profile)
mess_data_list = await self.host.memory.historyGet(
room_jid,
client.jid.userhostJID(),
limit=1,
between=True,
profile=client.profile
)
if mess_data_list:
timestamp = mess_data_list[0][1]
# we use seconds since last message to get backlog without duplicates
@ -994,27 +1010,27 @@ class SatMUCClient(muc.MUCClient):
else:
seconds = None
room = yield super(SatMUCClient, self).join(
room = await super(LiberviaMUCClient, self).join(
room_jid, nick, muc.HistoryOptions(seconds=seconds), password)
# used to send bridge signal once backlog are written in history
room._history_type = HISTORY_LEGACY
room._history_d = defer.Deferred()
room._history_d.callback(None)
defer.returnValue(room)
return room
@defer.inlineCallbacks
def _joinMAM(self, client, room_jid, nick, password):
"""Join room and retrieve history using MAM"""
room = yield super(SatMUCClient, self).join(
# we don't want any history from room as we'll get it with MAM
room_jid, nick, muc.HistoryOptions(maxStanzas=0), password=password)
room._history_type = HISTORY_MAM
async def _get_MAM_history(
self,
client: SatXMPPEntity,
room: muc.Room,
room_jid: jid.JID
) -> None:
"""Retrieve history for rooms handling MAM"""
history_d = room._history_d = defer.Deferred()
# we trigger now the deferred so all callback are processed as soon as possible
# and in order
history_d.callback(None)
last_mess = yield self.host.memory.historyGet(
last_mess = await self.host.memory.historyGet(
room_jid,
None,
limit=1,
@ -1040,7 +1056,7 @@ class SatMUCClient(muc.MUCClient):
count = 0
while not complete:
try:
mam_data = yield self._mam.getArchives(client, mam_req,
mam_data = await self._mam.getArchives(client, mam_req,
service=room_jid)
except xmpp_error.StanzaError as e:
if last_mess and e.condition == 'item-not-found':
@ -1107,20 +1123,35 @@ class SatMUCClient(muc.MUCClient):
errbackArgs=[room])
# we wait for all callbacks to be processed
yield history_d
await history_d
defer.returnValue(room)
async def _join_MAM(
self,
client: SatXMPPEntity,
room_jid: jid.JID,
nick: str,
password: Optional[str]
) -> muc.Room:
"""Join room and retrieve history using MAM"""
room = await super(LiberviaMUCClient, self).join(
# we don't want any history from room as we'll get it with MAM
room_jid, nick, muc.HistoryOptions(maxStanzas=0), password=password
)
room._history_type = HISTORY_MAM
# MAM history retrieval can be very long, and doesn't need to be sync, so we don't
# wait for it
defer.ensureDeferred(self._get_MAM_history(client, room, room_jid))
room.fully_joined.callback(room)
@defer.inlineCallbacks
def join(self, room_jid, nick, password=None):
return room
async def join(self, room_jid, nick, password=None):
room_service = jid.JID(room_jid.host)
has_mam = yield self.host.hasFeature(self.client, mam.NS_MAM, room_service)
has_mam = await self.host.hasFeature(self.client, mam.NS_MAM, room_service)
if not self._mam or not has_mam:
room = yield self._joinLegacy(self.client, room_jid, nick, password)
defer.returnValue(room)
return await self._join_legacy(self.client, room_jid, nick, password)
else:
room = yield self._joinMAM(self.client, room_jid, nick, password)
defer.returnValue(room)
return await self._join_MAM(self.client, room_jid, nick, password)
## presence/roster ##
@ -1400,13 +1431,14 @@ class SatMUCClient(muc.MUCClient):
this method will finish joining by:
- sending message to bridge
- calling fully_joined deferred
- calling fully_joined deferred (for legacy history)
- sending stanza put in cache
- cleaning variables not needed anymore
"""
args = self.plugin_parent._getRoomJoinedArgs(room, self.client.profile)
self.host.bridge.mucRoomJoined(*args)
room.fully_joined.callback(room)
if room._history_type == HISTORY_LEGACY:
room.fully_joined.callback(room)
del room._history_d
del room._history_type
cache = room._cache

11
sat/plugins/plugin_xep_0048.py
View file

@ -101,16 +101,15 @@ class XEP_0048(object):
except KeyError:
log.info(_("Text commands not available"))
@defer.inlineCallbacks
def profileConnected(self, client):
async def profileConnected(self, client):
local = client.bookmarks_local = PersistentBinaryDict(
NS_BOOKMARKS, client.profile
)
yield local.load()
await local.load()
if not local:
local[XEP_0048.MUC_TYPE] = dict()
local[XEP_0048.URL_TYPE] = dict()
private = yield self._getServerBookmarks("private", client.profile)
private = await self._getServerBookmarks("private", client.profile)
pubsub = client.bookmarks_pubsub = None
for bookmarks in (local, private, pubsub):
@ -118,7 +117,9 @@ class XEP_0048(object):
for (room_jid, data) in list(bookmarks[XEP_0048.MUC_TYPE].items()):
if data.get("autojoin", "false") == "true":
nick = data.get("nick", client.jid.user)
self.host.plugins["XEP-0045"].join(client, room_jid, nick, {})
defer.ensureDeferred(
self.host.plugins["XEP-0045"].join(client, room_jid, nick, {})
)
# we don't use a DeferredList to gather result here, as waiting for all room would
# slow down a lot the connection process, and result in a bad user experience.

4
sat/plugins/plugin_xep_0198.py
View file

@ -454,7 +454,9 @@ class XEP_0198(object):
if plg_0045 is not None:
# we re-join MUC rooms
muc_d_list = defer.DeferredList(
[plg_0045.join(*args) for args in muc_join_args])
[defer.ensureDeferred(plg_0045.join(*args))
for args in muc_join_args]
)
d.addCallback(lambda __: muc_d_list)
# at the end we replay the buffer, as those stanzas have probably not
# been received

25
sat/plugins/plugin_xep_0249.py
View file

@ -17,19 +17,20 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from sat.core.i18n import _, D_
from sat.core.constants import Const as C
from sat.core import exceptions
from sat.core.log import getLogger
log = getLogger(__name__)
from sat.tools import xml_tools
from twisted.words.xish import domish
from twisted.internet import defer
from twisted.words.protocols.jabber import jid
from twisted.words.xish import domish
from wokkel import disco, iwokkel
from zope.interface import implementer
from wokkel import disco, iwokkel
from sat.core import exceptions
from sat.core.constants import Const as C
from sat.core.i18n import D_, _
from sat.core.log import getLogger
from sat.tools import xml_tools
log = getLogger(__name__)
try:
@ -140,7 +141,9 @@ class XEP_0249(object):
_("Invitation accepted for room %(room)s [%(profile)s]")
% {"room": room_jid.userhost(), "profile": client.profile}
)
d = self.host.plugins["XEP-0045"].join(client, room_jid, client.jid.user, {})
d = defer.ensureDeferred(
self.host.plugins["XEP-0045"].join(client, room_jid, client.jid.user, {})
)
return d
def _messageReceivedTrigger(self, client, message_elt, post_treat):