core (xmpp): Return as soon as possible once connection is establied:
To improve connection time and UX in frontends, the `start_connection` returns immediately after the connection is established (authentication successful), and the rest of initialisation is now done in background. The `profile_ready` and `profile_connection_error` signals can be used to track the end of the full connection process. Roster retrieval is now also done in background for the same reason. `LiberviaRosterProtocol.got_roster` Deferred can be used to track roster retrieval.
This commit is contained in:
parent
8780d2133f
commit
45395c729a
2 changed files with 77 additions and 25 deletions
|
|
@ -1,7 +1,7 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
# Libervia types
|
||||
# Copyright (C) 2011 Jérôme Poisson (goffi@goffi.org)
|
||||
# Copyright (C) 2011-2026 Jérôme Poisson (goffi@goffi.org)
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
|
|
@ -18,7 +18,7 @@
|
|||
|
||||
from abc import ABC, abstractmethod
|
||||
from collections import namedtuple
|
||||
from typing import Any, TYPE_CHECKING, Type, TypeVar
|
||||
from typing import Any, Self, TYPE_CHECKING, Type, TypeVar
|
||||
|
||||
from twisted.internet import defer
|
||||
from twisted.python import failure
|
||||
|
|
@ -32,6 +32,7 @@ from libervia.backend.core import exceptions
|
|||
from libervia.backend.models.core import MessageData
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from libervia.backend.core.main import LiberviaBackend
|
||||
from libervia.backend.core.xmpp import (
|
||||
LiberviaRosterProtocol,
|
||||
SatDiscoProtocol,
|
||||
|
|
@ -150,7 +151,16 @@ class SatXMPPEntity(ABC):
|
|||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
async def start_connection(cls, host, profile, max_retries):
|
||||
async def start_connection(
|
||||
cls, host: "LiberviaBackend", profile: str, max_retries: int
|
||||
) -> Self | None:
|
||||
"""Start the XMPP connection for a profile.
|
||||
|
||||
@param host: LiberviaBackend instance.
|
||||
@param profile: name of the profile to connect.
|
||||
@param max_retries: maximum number of reconnection attempts.
|
||||
@return: the connected entity, or None if the connection was cancelled.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ import mimetypes
|
|||
from pathlib import Path
|
||||
import sys
|
||||
import time
|
||||
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, cast
|
||||
from typing import TYPE_CHECKING, Any, Dict, Optional, Self, Tuple, cast
|
||||
from urllib.parse import unquote, urlparse
|
||||
import uuid
|
||||
|
||||
|
|
@ -68,6 +68,8 @@ if TYPE_CHECKING:
|
|||
import txtorcon
|
||||
|
||||
from txtorcon import Tor
|
||||
|
||||
from libervia.backend.core.main import LiberviaBackend
|
||||
from txtorcon.util import default_control_port
|
||||
else:
|
||||
try:
|
||||
|
|
@ -196,13 +198,13 @@ class SatXMPPEntity(core_types.SatXMPPEntity):
|
|||
|
||||
## initialisation ##
|
||||
|
||||
async def _call_connection_triggers(self, connection_timer):
|
||||
"""Call conneting trigger prepare connected trigger
|
||||
async def _call_connection_triggers(
|
||||
self, connection_timer: dict[object, dict[str, float]]
|
||||
) -> list[tuple[object, Callable[..., Any]]]:
|
||||
"""Call connecting trigger prepare connected trigger.
|
||||
|
||||
@param plugins(iterable): plugins to use
|
||||
@return (list[object, callable]): plugin to trigger tuples with:
|
||||
- plugin instance
|
||||
- profile_connected* triggers (to call after connection)
|
||||
@param connection_timer: plugin connection timing data (mutated in-place).
|
||||
@return: list of (plugin instance, profile_connected callback) tuples.
|
||||
"""
|
||||
plugin_conn_cb = []
|
||||
self.plugins = plugins = self._get_plugins_list()
|
||||
|
|
@ -286,8 +288,22 @@ class SatXMPPEntity(core_types.SatXMPPEntity):
|
|||
raise exceptions.InitError(msg)
|
||||
|
||||
@classmethod
|
||||
async def start_connection(cls, host, profile, max_retries):
|
||||
"""instantiate the entity and start the connection"""
|
||||
async def start_connection(
|
||||
cls,
|
||||
host: "LiberviaBackend",
|
||||
profile: str,
|
||||
max_retries: int,
|
||||
) -> Self:
|
||||
"""Instantiate the entity and start the XMPP connection.
|
||||
|
||||
@param host: LiberviaBackend instance.
|
||||
@param profile: name of the profile to connect.
|
||||
@param max_retries: maximum number of reconnection attempts before giving up.
|
||||
|
||||
@return: the connected entity as soon as the XMPP stream is established.
|
||||
Post-connection setup (plugins profile_connected, profile_ready signal)
|
||||
continues in the background.
|
||||
"""
|
||||
# FIXME: reconnection doesn't seems to be handled correclty
|
||||
# (client is deleted then recreated from scratch)
|
||||
# most of methods called here should be called once on first connection
|
||||
|
|
@ -388,14 +404,35 @@ class SatXMPPEntity(core_types.SatXMPPEntity):
|
|||
|
||||
log.debug(_("setting plugins parents"))
|
||||
|
||||
connection_timer: Dict[str, Dict[str, float]] = {}
|
||||
connection_timer: dict[object, dict[str, float]] = {}
|
||||
plugin_conn_cb = await entity._call_connection_triggers(connection_timer)
|
||||
|
||||
entity.startService()
|
||||
|
||||
await entity.conn_deferred
|
||||
|
||||
await defer.maybeDeferred(entity.entity_connected)
|
||||
# Connection is established we return the entity immediately.
|
||||
defer.ensureDeferred(
|
||||
entity._post_connection(profile, connection_timer, plugin_conn_cb)
|
||||
)
|
||||
return entity
|
||||
finally:
|
||||
cls.profiles_connecting.remove(profile)
|
||||
|
||||
async def _post_connection(
|
||||
self,
|
||||
profile: str,
|
||||
connection_timer: dict[object, dict[str, float]],
|
||||
plugin_conn_cb: list[tuple[object, Callable[..., Any]]],
|
||||
) -> None:
|
||||
"""Post-connection setup.
|
||||
|
||||
Called in background after connection is esablished, and stream is running.
|
||||
@param profile: Name of the profile being connected.
|
||||
@param connection_timer: plugin connection timing data.
|
||||
@param plugin_conn_cb: list of (plugin, profile_connected callback) tuples.
|
||||
"""
|
||||
try:
|
||||
await defer.maybeDeferred(self.entity_connected)
|
||||
|
||||
# Call profile_connected callback for all plugins,
|
||||
# and print error message if any of them fails
|
||||
|
|
@ -403,8 +440,8 @@ class SatXMPPEntity(core_types.SatXMPPEntity):
|
|||
for plugin, callback in plugin_conn_cb:
|
||||
conn_cb_list.append(
|
||||
defer.ensureDeferred(
|
||||
cls._run_profile_connected(
|
||||
callback, entity, connection_timer[plugin]
|
||||
self._run_profile_connected(
|
||||
callback, self, connection_timer[plugin]
|
||||
)
|
||||
)
|
||||
)
|
||||
|
|
@ -433,9 +470,8 @@ class SatXMPPEntity(core_types.SatXMPPEntity):
|
|||
# DeferredList
|
||||
total_plugins = 0
|
||||
# total real sum all connecting (which happen sequentially) and the
|
||||
# longuest connected (connected happen in parallel, thus the longuest is
|
||||
# longest connected (connected happen in parallel, thus the longest is
|
||||
# roughly the total time for connected)
|
||||
total_real = 0
|
||||
total_real = max(t.get("connected", 0) for t in connection_timer.values())
|
||||
|
||||
for plugin in plugins_by_timer:
|
||||
|
|
@ -463,10 +499,12 @@ class SatXMPPEntity(core_types.SatXMPPEntity):
|
|||
list_d.addCallback(log_plugin_results)
|
||||
# FIXME: we should have a timeout here, and a way to know if a plugin freeze
|
||||
await list_d
|
||||
except BaseException as e:
|
||||
log.exception(f"Connection setup failed for {self.profile!r}: {e}")
|
||||
self.host_app.bridge.profile_connection_error(str(e), self.profile)
|
||||
raise
|
||||
|
||||
host.bridge.profile_ready(profile)
|
||||
finally:
|
||||
cls.profiles_connecting.remove(profile)
|
||||
self.host_app.bridge.profile_ready(profile)
|
||||
|
||||
def _disconnection_cb(self, __):
|
||||
self._connected_d = None
|
||||
|
|
@ -1151,7 +1189,9 @@ class SatXMPPClient(SatXMPPEntity, core_types.SatXMPPClient, tx_xmpp_client.XMPP
|
|||
return super()._getConnection()
|
||||
|
||||
@classmethod
|
||||
async def start_connection(cls, host, profile, max_retries):
|
||||
async def start_connection(
|
||||
cls, host: "LiberviaBackend", profile: str, max_retries: int
|
||||
) -> None:
|
||||
try:
|
||||
await super(SatXMPPClient, cls).start_connection(host, profile, max_retries)
|
||||
except exceptions.CancelError as e:
|
||||
|
|
@ -1199,8 +1239,10 @@ class SatXMPPClient(SatXMPPEntity, core_types.SatXMPPClient, tx_xmpp_client.XMPP
|
|||
)
|
||||
|
||||
def _finish_connection(self, __):
|
||||
d = defer.ensureDeferred(self.roster.request_roster())
|
||||
d.addCallback(lambda __: super(SatXMPPClient, self)._finish_connection(__))
|
||||
# We run "request_roster" in background because we want to return as soon as
|
||||
# possible once connection is established, to have good UX in frontends.
|
||||
defer.ensureDeferred(self.roster.request_roster())
|
||||
super(SatXMPPClient, self)._finish_connection(__)
|
||||
|
||||
|
||||
@implementer(itx_xmpp.IDisco)
|
||||
|
|
|
|||
Loading…
Reference in a new issue