Mercurial > libervia-backend
changeset 3823:5d72dc52ee4a
plugin XEP-0163: allow unset `event_type` in `addPEPEvent`, check conflict + type hints
author | Goffi <goffi@goffi.org> |
---|---|
date | Wed, 29 Jun 2022 12:08:53 +0200 |
parents | 65bac82e4049 |
children | 6329ee6b6df4 |
files | sat/plugins/plugin_xep_0163.py |
diffstat | 1 files changed, 25 insertions(+), 9 deletions(-) [+] |
line wrap: on
line diff
--- a/sat/plugins/plugin_xep_0163.py Wed Jun 29 12:07:45 2022 +0200 +++ b/sat/plugins/plugin_xep_0163.py Wed Jun 29 12:08:53 2022 +0200 @@ -17,18 +17,21 @@ # 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 typing import Optional, Callable from sat.core.i18n import _ from sat.core import exceptions from sat.core.constants import Const as C from sat.core.log import getLogger -log = getLogger(__name__) from twisted.words.xish import domish from wokkel import disco, pubsub from wokkel.formats import Mood from sat.tools.common import data_format + +log = getLogger(__name__) + NS_USER_MOOD = "http://jabber.org/protocol/mood" PLUGIN_INFO = { @@ -71,21 +74,34 @@ disco_info.extend(list(map(disco.DiscoFeature, self.pep_events))) return True - def addPEPEvent(self, event_type, node, in_callback, out_callback=None, notify=True): + def addPEPEvent( + self, + event_type: Optional[str], + node: str, + in_callback: Callable, + out_callback: Optional[Callable] = None, + notify: bool = True + ) -> None: """Add a Personal Eventing Protocol event manager - @param event_type(unicode): type of the event (always uppercase), - can be MOOD, TUNE, etc - @param node(unicode): namespace of the node (e.g. http://jabber.org/protocol/mood + @param event_type: type of the event (stored uppercase), + only used when out_callback is set. + Can be MOOD, TUNE, etc. + @param node: namespace of the node (e.g. http://jabber.org/protocol/mood for User Mood) - @param in_callback(callable): method to call when this event occur + @param in_callback: method to call when this event occur the callable will be called with (itemsEvent, profile) as arguments - @param out_callback(callable,None): method to call when we want to publish this + @param out_callback: method to call when we want to publish this event (must return a deferred) the callable will be called when sendPEPEvent is called - @param notify(bool): add autosubscribe (+notify) if True + @param notify: add autosubscribe (+notify) if True """ - if out_callback: + if event_type and out_callback: + event_type = event_type.upper() + if event_type in self.pep_out_cb: + raise exceptions.ConflictError( + f"event_type {event_type!r} already exists" + ) self.pep_out_cb[event_type] = out_callback self.pep_events.add(node) if notify: