Mercurial > libervia-backend
diff libervia/frontends/quick_frontend/quick_app.py @ 4126:45e3bb8607d8
frontends (quick app): allow frontend to register an action handler:
Some feature may need to handle directly some action, such as the incoming `A/V call`
feature.
rel 424
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 03 Oct 2023 16:25:25 +0200 |
parents | bb74f7dc3b10 |
children | d01b8d002619 |
line wrap: on
line diff
--- a/libervia/frontends/quick_frontend/quick_app.py Tue Oct 03 16:24:13 2023 +0200 +++ b/libervia/frontends/quick_frontend/quick_app.py Tue Oct 03 16:25:25 2023 +0200 @@ -16,22 +16,22 @@ # 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 libervia.backend.core.log import getLogger +import sys +import time +from typing import Callable + +from libervia.backend.core import exceptions from libervia.backend.core.i18n import _ -from libervia.backend.core import exceptions +from libervia.backend.core.log import getLogger from libervia.backend.tools import trigger from libervia.backend.tools.common import data_format - -from libervia.frontends.tools import jid from libervia.frontends.quick_frontend import quick_widgets from libervia.frontends.quick_frontend import quick_menus from libervia.frontends.quick_frontend import quick_blog from libervia.frontends.quick_frontend import quick_chat, quick_games from libervia.frontends.quick_frontend import quick_contact_list from libervia.frontends.quick_frontend.constants import Const as C - -import sys -import time +from libervia.frontends.tools import jid log = getLogger(__name__) @@ -291,6 +291,9 @@ self._listeners = {} # key: listener type ("avatar", "selected", etc), # value: list of callbacks + # cf. [register_action_handler] + self._action_handlers: dict[str, Callable[[dict, str, int, str], None]] = {} + # triggers self.trigger = ( trigger.TriggerManager() @@ -752,10 +755,16 @@ self.sync = False self.set_presence_status(C.PRESENCE_UNAVAILABLE, "", profile=profile) - def action_new_handler(self, action_data_s, id_, security_limit, profile): - self.action_manager( - data_format.deserialise(action_data_s), user_action=False, profile=profile - ) + def action_new_handler(self, action_data_s, action_id, security_limit, profile): + action_data = data_format.deserialise(action_data_s) + action_type = action_data.get("type") + action_handler = self._action_handlers.get(action_type) + if action_handler is not None: + action_handler(action_data, action_id, security_limit, profile) + else: + self.action_manager( + action_data, user_action=False, profile=profile + ) def contact_new_handler(self, jid_s, attributes, groups, profile): entity = jid.JID(jid_s) @@ -1266,6 +1275,35 @@ self.contact_lists[profile].set_cache(entity, "avatar", value) self.call_listeners("avatar", entity, value, profile=profile) + def register_action_handler( + self, + action_type: str, + handler: Callable[[dict, str, int, str], None] + ) -> None: + """Register a handler for action type. + + If an action of this type is received, the handler will be used, otherwise, + generic ``action_manager`` is used + @param action_type: type of action that the handler manage + @param handler: method to call when an actipn of ``action_type`` is received. + Will have following args: + ``action_data`` + Data of the action. + ``action_id`` + ID of the action. + ``security_limit`` + Security limit, used to check if this action can be used in current security + context. + ``profile`` + Profile name. + """ + if handler in self._action_handlers: + raise exceptions.ConflictError( + f"There is already a registered handler for {action_type} actions: " + f"{handler}" + ) + self._action_handlers[action_type] = handler + def action_manager(self, action_data, callback=None, ui_show_cb=None, user_action=True, progress_cb=None, progress_eb=None, profile=C.PROF_KEY_NONE): """Handle backend action