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
This commit is contained in:
Goffi 2023-10-03 16:25:25 +02:00
commit 366f827beb

View file

@ -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
from libervia.backend.core.i18n import _
import sys
import time
from typing import Callable
from libervia.backend.core import exceptions
from libervia.backend.core.i18n import _
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 @@ class QuickApp:
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 @@ class QuickApp:
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 @@ class QuickApp:
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