Philipp Hörist pushed to branch tasks at gajim / python-nbxmpp

Commits: 36aef2b6 by wurstsalat at 2020-08-29T19:41:27+02:00 DOAP: Add XEP-0106

- - - - - df93bf7d by lovetox at 2020-09-16T13:51:05+02:00 Return copy on get_bound_jid()

- - - - - a09e3e7c by lovetox at 2020-09-16T13:51:05+02:00 Add Task module

Module provides a decorator which wraps a generator in a Task The Task iterates the generator until it ends and processes the yielded values.

Example: ``` @run_as_task request_something(self, jid): task = yield

response = yield get_request_iq() result = process(response)

if is_error_result(result): raise StanzaError(result) yield result ```

- - - - - e72c3edc by lovetox at 2020-09-16T13:51:05+02:00 Add util module

- - - - - 66e08be9 by lovetox at 2020-09-16T13:51:05+02:00 Protocol: Add isError() and isResult() methods

- - - - - 185dbf55 by lovetox at 2020-09-16T13:51:05+02:00 Add Namespace

- - - - - ab166ef0 by lovetox at 2020-09-16T13:51:05+02:00 Add error module

- - - - - 8d3a3653 by lovetox at 2020-09-16T13:51:05+02:00 Add VCard Temp (XEP-0054) support

- - - - - 928ce0a7 by lovetox at 2020-09-16T13:51:05+02:00 UserAvatar: Add set_avatar()

- - - - - 9b387cc7 by lovetox at 2020-09-16T13:51:05+02:00 PubSub: Rewrite and use Tasks

- - - - - b4a9d548 by lovetox at 2020-09-16T13:51:05+02:00 user_avatar

- - - - -

21 changed files:

- nbxmpp/client.py - nbxmpp/dispatcher.py - + nbxmpp/errors.py - nbxmpp/modules/activity.py - nbxmpp/modules/base.py - nbxmpp/modules/bookmarks.py - nbxmpp/modules/location.py - nbxmpp/modules/mood.py - nbxmpp/modules/nickname.py - nbxmpp/modules/omemo.py - nbxmpp/modules/openpgp.py - nbxmpp/modules/pubsub.py - nbxmpp/modules/tune.py - nbxmpp/modules/user_avatar.py - + nbxmpp/modules/util.py - + nbxmpp/modules/vcard_temp.py - nbxmpp/namespaces.py - nbxmpp/protocol.py - nbxmpp/structs.py - + nbxmpp/task.py - python-nbxmpp.doap

Changes:

===================================== nbxmpp/client.py ===================================== @@ -119,6 +119,7 @@ class Client(Observable): self._mode = Mode.CLIENT

self._ping_source_id = None + self._tasks = []

self._dispatcher = StanzaDispatcher(self) self._dispatcher.subscribe('before-dispatch', self._on_before_dispatch) @@ -130,6 +131,15 @@ class Client(Observable):

self._state = StreamState.DISCONNECTED

+ def add_task(self, task): + self._tasks.append(task) + + def remove_task(self, task, _context): + try: + self._tasks.remove(task) + except Exception: + pass + @property def log_context(self): return self._log_context @@ -310,7 +320,9 @@ class Client(Observable): return self._proxy

def get_bound_jid(self): - return self._jid + if self._jid is None: + return None + return self._jid.copy()

def _set_bound_jid(self, jid): self._jid = JID.from_string(jid) @@ -447,6 +459,8 @@ class Client(Observable):

def _on_disconnected(self, _connection, _signal_name): self.state = StreamState.DISCONNECTED + for task in self._tasks: + task.cancel() self._remove_ping_timer() self._dispatcher.remove_ping_callback(self._ping_id) self._reset_stream() @@ -833,6 +847,8 @@ class Client(Observable): self._dispatcher.unregister_handler(*args, **kwargs)

def destroy(self): + for task in self._tasks: + task.cancel() self._remove_ping_timer() self._smacks = None self._sasl = None

===================================== nbxmpp/dispatcher.py ===================================== @@ -75,6 +75,7 @@ from nbxmpp.modules.chatstates import Chatstates from nbxmpp.modules.register import Register from nbxmpp.modules.http_upload import HTTPUpload from nbxmpp.modules.mam import MAM +from nbxmpp.modules.vcard_temp import VCardTemp from nbxmpp.modules.misc import unwrap_carbon from nbxmpp.modules.misc import unwrap_mam from nbxmpp.structs import StanzaTimeoutError @@ -180,6 +181,7 @@ class StanzaDispatcher(Observable): self._modules['Register'] = Register(self._client) self._modules['HTTPUpload'] = HTTPUpload(self._client) self._modules['MAM'] = MAM(self._client) + self._modules['VCardTemp'] = VCardTemp(self._client)

for instance in self._modules.values(): for handler in instance.handlers:

===================================== nbxmpp/errors.py ===================================== @@ -0,0 +1,106 @@ +# Copyright (C) 2020 Philipp Hörist <philipp AT hoerist.com> +# +# This file is part of nbxmpp. +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; either version 3 +# of the License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; If not, see http://www.gnu.org/licenses/. + +from nbxmpp.namespaces import Namespace + + +def is_error(error): + return isinstance(error, BaseError) + + +class BaseError(Exception): + def __init__(self, is_fatal=False): + self.is_fatal = is_fatal + + +class StanzaError(BaseError): + + log_level = 'info' + app_namespace = None + + def __init__(self, stanza): + BaseError.__init__(self) + self.stanza = stanza + self._stanza_name = stanza.getName() + self._error_node = stanza.getTag('error') + self.condition = stanza.getError() + self.condition_data = self._error_node.getTagData(self.condition) + self.app_condition = self._get_app_condition() + self.type = stanza.getErrorType() + self.jid = stanza.getFrom() + self.id = stanza.getID() + self._text = {} + + text_elements = self._error_node.getTags('text', + namespace=Namespace.STANZAS) + for element in text_elements: + lang = element.getXmlLang() + text = element.getData() + self._text[lang] = text + + def _get_app_condition(self): + if self.app_namespace is None: + return None + + for node in self._error_node.getChildren(): + if node.getNamespace() == self.app_namespace: + return node.getName() + return None + + def get_text(self, pref_lang=None): + if pref_lang is not None: + text = self._text.get(pref_lang) + if text is not None: + return text + + if self._text: + text = self._text.get('en') + if text is not None: + return text + + text = self._text.get(None) + if text is not None: + return text + return self._text.popitem()[1] + return '' + + def set_text(self, lang, text): + self._text[lang] = text + + def __str__(self): + condition = self.condition + if self.app_condition is not None: + condition = '%s (%s)' % (self.condition, self.app_condition) + text = self.get_text('en') or '' + if text: + text = ' - %s' % text + return 'Error from %s: %s%s' % (self.jid, condition, text) + + +class PubSubStanzaError(StanzaError): + + app_namespace = Namespace.PUBSUB_ERROR + + +class MalformedStanzaError(BaseError): + + log_level = 'warning' + + def __init__(self, text, stanza, is_fatal=True): + BaseError.__init__(self, is_fatal=is_fatal) + self.stanza = stanza + self.text = str(text)

===================================== nbxmpp/modules/activity.py ===================================== @@ -25,6 +25,11 @@ from nbxmpp.modules.base import BaseModule

class Activity(BaseModule): + + _depends = { + 'publish': 'PubSub' + } + def __init__(self, client): BaseModule.__init__(self, client)

@@ -91,6 +96,5 @@ class Activity(BaseModule): if data.text: item.addChild('text', payload=data.text)

- jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').publish( - jid, Namespace.ACTIVITY, item, id_='current') + task = self.publish(Namespace.ACTIVITY, item, id_='current') + task.start()

===================================== nbxmpp/modules/base.py ===================================== @@ -21,7 +21,17 @@ from nbxmpp.util import LogAdapter

class BaseModule: + + _depends = {} + def __init__(self, client): logger_name = 'nbxmpp.m.%s' % self.__class__.__name__.lower() self._log = LogAdapter(logging.getLogger(logger_name), {'context': client.log_context}) + + def __getattr__(self, name): + if name not in self._depends: + raise AttributeError + + module = self._client.get_module(self._depends[name]) + return getattr(module, name)

===================================== nbxmpp/modules/bookmarks.py ===================================== @@ -30,11 +30,9 @@ from nbxmpp.util import to_xs_boolean from nbxmpp.util import call_on_response from nbxmpp.util import callback from nbxmpp.util import raise_error -from nbxmpp.util import is_error_result from nbxmpp.modules.pubsub import get_pubsub_item from nbxmpp.modules.pubsub import get_pubsub_items from nbxmpp.modules.pubsub import get_pubsub_request -from nbxmpp.modules.pubsub import get_publish_options from nbxmpp.modules.base import BaseModule

@@ -47,13 +45,19 @@ BOOKMARK_2_OPTIONS = { 'pubsub#notify_delete': 'true', 'pubsub#notify_retract': 'true', 'pubsub#persist_items': 'true', - 'pubsub#max_items': '128', + 'pubsub#max_items': 'max', 'pubsub#access_model': 'whitelist', 'pubsub#send_last_published_item': 'never', }

class Bookmarks(BaseModule): + + _depends = { + 'retract': 'PubSub', + 'publish': 'PubSub', + } + def __init__(self, client): BaseModule.__init__(self, client)

@@ -71,8 +75,6 @@ class Bookmarks(BaseModule):

self._bookmark_2_queue = {} self._bookmark_1_queue = [] - self._node_configuration_in_progress = False - self._node_configuration_not_possible = False

def _process_pubsub_bookmarks(self, _client, stanza, properties): if not properties.is_pubsub_event: @@ -315,115 +317,45 @@ class Bookmarks(BaseModule):

def retract_bookmark(self, bookmark_jid): self._log.info('Retract Bookmark: %s', bookmark_jid) - jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').retract(jid, - Namespace.BOOKMARKS_2, - str(bookmark_jid)) + + task = self.retract(Namespace.BOOKMARKS_2, str(bookmark_jid)) + task.start()

def _store_bookmark_1(self, bookmarks): self._log.info('Store Bookmarks 1 (PubSub)') - jid = self._client.get_bound_jid().bare + self._bookmark_1_queue = bookmarks item = self._build_storage_node(bookmarks) - options = get_publish_options(BOOKMARK_1_OPTIONS) - self._client.get_module('PubSub').publish( - jid, - Namespace.BOOKMARKS, - item, - id_='current', - options=options, - callback=self._on_store_bookmark_result, - user_data=Namespace.BOOKMARKS) + self.publish(Namespace.BOOKMARKS, + item, + id_='current', + options=BOOKMARK_1_OPTIONS, + force_node_options=True, + callback=self._on_store_bookmark_result, + user_data=Namespace.BOOKMARKS)

def _store_bookmark_2(self, bookmarks): - if self._node_configuration_not_possible: - self._log.warning('Node configuration not possible') - return - self._log.info('Store Bookmarks 2 (PubSub)') - jid = self._client.get_bound_jid().bare + for bookmark in bookmarks: self._bookmark_2_queue[bookmark.jid] = bookmark item = self._build_conference_node(bookmark) - options = get_publish_options(BOOKMARK_2_OPTIONS) - self._client.get_module('PubSub').publish( - jid, - Namespace.BOOKMARKS_2, - item, - id_=str(bookmark.jid), - options=options, - callback=self._on_store_bookmark_result, - user_data=Namespace.BOOKMARKS_2) - - def _on_store_bookmark_result(self, result, node): - if not is_error_result(result): - self._bookmark_1_queue = [] - self._bookmark_2_queue.pop(result.id, None) - return - - if (result.condition == 'conflict' and - result.app_condition == 'precondition-not-met'): - if self._node_configuration_in_progress: - return - - self._node_configuration_in_progress = True - jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').get_node_configuration( - jid, - node, - callback=self._on_node_configuration_received) - - else: - self._bookmark_1_queue = [] - self._bookmark_2_queue.pop(result.id, None) - self._log.warning(result) - - def _on_node_configuration_received(self, result): - if is_error_result(result): - self._log.warning(result) - self._bookmark_1_queue = [] - self._bookmark_2_queue.clear() - return - - if result.node == Namespace.BOOKMARKS: - config = BOOKMARK_1_OPTIONS - else: - config = BOOKMARK_2_OPTIONS - self._apply_config(result.form, config) - self._client.get_module('PubSub').set_node_configuration( - result.jid, - result.node, - result.form, - callback=self._on_node_configuration_finished) - - def _on_node_configuration_finished(self, result): - self._node_configuration_in_progress = False - if is_error_result(result): - self._log.warning(result) - self._bookmark_2_queue.clear() - self._bookmark_1_queue = [] - self._node_configuration_not_possible = True - return - - self._log.info('Republish bookmarks') - if self._bookmark_2_queue: - bookmarks = self._bookmark_2_queue.copy() - self._bookmark_2_queue.clear() - self._store_bookmark_2(bookmarks.values()) - else: - bookmarks = self._bookmark_1_queue.copy() - self._bookmark_1_queue.clear() - self._store_bookmark_1(bookmarks) + self.publish(Namespace.BOOKMARKS_2, + item, + id_=str(bookmark.jid), + options=BOOKMARK_2_OPTIONS, + force_node_options=True, + callback=self._on_store_bookmark_result, + user_data=Namespace.BOOKMARKS_2) + + def _on_store_bookmark_result(self, task): + try: + result = task.finish() + except Exception as error: + self._log.warning(error)

- @staticmethod - def _apply_config(form, config): - for var, value in config.items(): - try: - field = form[var] - except KeyError: - pass - else: - field.value = value + self._bookmark_1_queue = [] + self._bookmark_2_queue.pop(result.id, None)

@call_on_response('_on_private_store_result') def _store_with_private(self, bookmarks):

===================================== nbxmpp/modules/location.py ===================================== @@ -24,6 +24,11 @@ from nbxmpp.modules.base import BaseModule

class Location(BaseModule): + + _depends = { + 'publish': 'PubSub' + } + def __init__(self, client): BaseModule.__init__(self, client)

@@ -70,6 +75,5 @@ class Location(BaseModule): if value is not None: item.addChild(tag, payload=value)

- jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').publish( - jid, Namespace.LOCATION, item, id_='current') + task = self.publish(Namespace.LOCATION, item, id_='current') + task.start()

===================================== nbxmpp/modules/mood.py ===================================== @@ -25,6 +25,11 @@ from nbxmpp.modules.base import BaseModule

class Mood(BaseModule): + + _depends = { + 'publish': 'PubSub' + } + def __init__(self, client): BaseModule.__init__(self, client)

@@ -80,6 +85,5 @@ class Mood(BaseModule): if data.text: item.addChild('text', payload=data.text)

- jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').publish( - jid, Namespace.MOOD, item, id_='current') + task = self.publish(Namespace.MOOD, item, id_='current') + task.start()

===================================== nbxmpp/modules/nickname.py ===================================== @@ -23,6 +23,11 @@ from nbxmpp.modules.base import BaseModule

class Nickname(BaseModule): + + _depends = { + 'publish': 'PubSub' + } + def __init__(self, client): BaseModule.__init__(self, client)

@@ -87,6 +92,6 @@ class Nickname(BaseModule): item = Node('nick', {'xmlns': Namespace.NICK}) if nickname is not None: item.addData(nickname) - jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').publish( - jid, Namespace.NICK, item, id_='current') + + task = self.publish(Namespace.NICK, item, id_='current') + task.start()

===================================== nbxmpp/modules/omemo.py ===================================== @@ -35,6 +35,11 @@ from nbxmpp.modules.base import BaseModule

class OMEMO(BaseModule): + + _depends = { + 'publish': 'PubSub' + } + def __init__(self, client): BaseModule.__init__(self, client)

@@ -192,9 +197,9 @@ class OMEMO(BaseModule): item.addChild('device').setAttr('id', device)

self._log.info('Set devicelist: %s', devicelist) - jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').publish( - jid, Namespace.OMEMO_TEMP_DL, item, id_='current') + + task = self.publish(Namespace.OMEMO_TEMP_DL, item, id_='current') + task.start()

@call_on_response('_devicelist_received') def request_devicelist(self, jid=None): @@ -225,9 +230,9 @@ class OMEMO(BaseModule): self._log.info('Set bundle')

node = '%s:%s' % (Namespace.OMEMO_TEMP_BUNDLE, device_id) - jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').publish( - jid, node, item, id_='current') + + task = self.publish(node, item, id_='current') + task.start()

@staticmethod def _create_bundle(bundle):

===================================== nbxmpp/modules/openpgp.py ===================================== @@ -38,6 +38,11 @@ from nbxmpp.modules.base import BaseModule

class OpenPGP(BaseModule): + + _depends = { + 'publish': 'PubSub' + } + def __init__(self, client): BaseModule.__init__(self, client)

@@ -154,9 +159,9 @@ class OpenPGP(BaseModule): item.addChild('pubkey-metadata', attrs=attrs)

self._log.info('Set keylist: %s', keylist) - jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').publish( - jid, Namespace.OPENPGP_PK, item, id_='current') + + task = self.publish(Namespace.OPENPGP_PK, item, id_='current') + task.start()

def set_public_key(self, key, fingerprint, date): date = time.strftime( @@ -168,9 +173,9 @@ class OpenPGP(BaseModule): node = '%s:%s' % (Namespace.OPENPGP_PK, fingerprint)

self._log.info('Set public key') - jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').publish( - jid, node, item, id_='current') + + task = self.publish(node, item, id_='current') + task.start()

@call_on_response('_public_key_received') def request_public_key(self, jid, fingerprint): @@ -274,9 +279,9 @@ class OpenPGP(BaseModule): item.setData(b64encode(secret_key))

self._log.info('Set secret key') - jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').publish( - jid, Namespace.OPENPGP_SK, item, id_='current') + + task = self.publish(Namespace.OPENPGP_SK, item, id_='current') + task.start()

def parse_signcrypt(stanza):

===================================== nbxmpp/modules/pubsub.py ===================================== @@ -1,4 +1,4 @@ -# Copyright (C) 2018 Philipp Hörist <philipp AT hoerist.com> +# Copyright (C) 2020 Philipp Hörist <philipp AT hoerist.com> # # This file is part of nbxmpp. # @@ -15,20 +15,20 @@ # You should have received a copy of the GNU General Public License # along with this program; If not, see http://www.gnu.org/licenses/.

-from nbxmpp.namespaces import Namespace -from nbxmpp.protocol import Node -from nbxmpp.protocol import Iq -from nbxmpp.protocol import isResultNode +from collections import namedtuple + +from nbxmpp.task import xmpp_task +from nbxmpp.errors import is_error +from nbxmpp.errors import PubSubStanzaError +from nbxmpp.errors import MalformedStanzaError from nbxmpp.structs import StanzaHandler from nbxmpp.structs import PubSubEventData -from nbxmpp.structs import CommonResult -from nbxmpp.structs import PubSubConfigResult -from nbxmpp.structs import PubSubPublishResult -from nbxmpp.modules.dataforms import extend_form -from nbxmpp.util import call_on_response -from nbxmpp.util import callback -from nbxmpp.util import raise_error +from nbxmpp.protocol import Iq +from nbxmpp.protocol import Node +from nbxmpp.namespaces import Namespace from nbxmpp.modules.base import BaseModule +from nbxmpp.modules.util import process_response +from nbxmpp.modules.dataforms import extend_form

class PubSub(BaseModule): @@ -84,107 +84,101 @@ class PubSub(BaseModule): id_ = item.getAttr('id') properties.pubsub_event = PubSubEventData(node, id_, item)

- @call_on_response('_publish_result_received') - def publish(self, jid, node, item, id_=None, options=None): - query = Iq('set', to=jid) - pubsub = query.addChild('pubsub', namespace=Namespace.PUBSUB) - publish = pubsub.addChild('publish', {'node': node}) - attrs = {} - if id_ is not None: - attrs = {'id': id_} - publish.addChild('item', attrs, [item]) - if options: - publish = pubsub.addChild('publish-options') - publish.addChild(node=options) - return query - - @callback - def _publish_result_received(self, stanza): - if not isResultNode(stanza): - return raise_error(self._log.warning, stanza) - - jid = stanza.getFrom() - pubsub = stanza.getTag('pubsub', namespace=Namespace.PUBSUB) - if pubsub is None: - # XEP-0060: IQ payload is not mandatory on result - return PubSubPublishResult(jid, None, None) - - publish = pubsub.getTag('publish') - if publish is None: - return raise_error(self._log.warning, stanza, 'stanza-malformed') - - node = publish.getAttr('node') - item = publish.getTag('item') - if item is None: - return raise_error(self._log.warning, stanza, 'stanza-malformed') - - id_ = item.getAttr('id') - return PubSubPublishResult(jid, node, id_) - - @call_on_response('_default_response') - def retract(self, jid, node, id_, notify=True): - query = Iq('set', to=jid) - pubsub = query.addChild('pubsub', namespace=Namespace.PUBSUB) - attrs = {'node': node} - if notify: - attrs['notify'] = 'true' - retract = pubsub.addChild('retract', attrs=attrs) - retract.addChild('item', {'id': id_}) - return query - - @call_on_response('_default_response') - def set_node_configuration(self, jid, node, form): - self._log.info('Set configuration for %s %s', node, jid) - query = Iq('set', to=jid) - pubsub = query.addChild('pubsub', namespace=Namespace.PUBSUB_OWNER) - configure = pubsub.addChild('configure', {'node': node}) - form.setAttr('type', 'submit') - configure.addChild(node=form) - return query - - @call_on_response('_node_configuration_received') - def get_node_configuration(self, jid, node): - self._log.info('Request node configuration') - query = Iq('get', to=jid) - pubsub = query.addChild('pubsub', namespace=Namespace.PUBSUB_OWNER) - pubsub.addChild('configure', {'node': node}) - return query - - @callback - def _node_configuration_received(self, stanza): - if not isResultNode(stanza): - return raise_error(self._log.warning, stanza) - - jid = stanza.getFrom() - pubsub = stanza.getTag('pubsub', namespace=Namespace.PUBSUB_OWNER) - if pubsub is None: - return raise_error(self._log.warning, stanza, 'stanza-malformed', - 'No pubsub node found') - - configure = pubsub.getTag('configure') - if configure is None: - return raise_error(self._log.warning, stanza, 'stanza-malformed', - 'No configure node found') - - node = configure.getAttr('node') - - forms = configure.getTags('x', namespace=Namespace.DATA) - for form in forms: - dataform = extend_form(node=form) - form_type = dataform.vars.get('FORM_TYPE') - if form_type is None or form_type.value != Namespace.PUBSUB_CONFIG: - continue - self._log.info('Node configuration received from: %s', jid) - return PubSubConfigResult(jid=jid, node=node, form=dataform) - - return raise_error(self._log.warning, stanza, 'stanza-malformed', - 'No valid form type found') - - @callback - def _default_response(self, stanza): - if not isResultNode(stanza): - return raise_error(self._log.info, stanza) - return CommonResult(jid=stanza.getFrom()) + @xmpp_task + def request_item(self, node, id_, jid=None): + task = yield + + response = yield _make_pubsub_request(node, id_=id_, jid=jid) + + if response.isError(): + raise PubSubStanzaError(response) + + item = _get_pubsub_item(response, node, id_) + yield task.set_result(item) + + @xmpp_task + def request_items(self, node, max_items=None, jid=None): + _task = yield + + response = yield _make_pubsub_request(node, + max_items=max_items, + jid=jid) + + if response.isError(): + raise PubSubStanzaError(response) + + yield _get_pubsub_items(response, node) + + @xmpp_task + def publish(self, + node, + item, + id_=None, + options=None, + jid=None, + force_node_options=False): + + _task = yield + + request = _make_publish_request(node, item, id_, options, jid) + response = yield request + + if response.isError(): + error = PubSubStanzaError(response) + if (not force_node_options or + error.app_condition != 'precondition-not-met'): + raise error + + result = yield self.reconfigure_node(node, options, jid) + if is_error(result): + raise result + + response = yield request + if response.isError(): + raise PubSubStanzaError(response) + + jid = response.getFrom() + item_id = _get_published_item_id(response, node, id_) + yield PubSubPublishResult(jid, node, item_id) + + @xmpp_task + def retract(self, node, id_, jid=None, notify=True): + _task = yield + + response = yield _make_retract_request(node, id_, jid, notify) + yield process_response(response) + + @xmpp_task + def reconfigure_node(self, node, options, jid=None): + _task = yield + + result = yield self.get_node_configuration(node, jid) + if is_error(result): + raise result + + _apply_options(result.form, options) + result = yield self.set_node_configuration(node, result.form, jid) + yield result + + @xmpp_task + def set_node_configuration(self, node, form, jid=None): + _task = yield + + response = yield _make_node_configuration(node, form, jid) + yield process_response(response) + + @xmpp_task + def get_node_configuration(self, node, jid=None): + _task = yield + + response = yield _make_node_configuration_request(node, jid) + + if response.isError(): + raise PubSubStanzaError(response) + + jid = response.getFrom() + form = _get_configure_form(response, node) + yield PubSubNodeConfigurationResult(jid=jid, node=node, form=form)

def get_pubsub_request(jid, node, id_=None, max_items=None): @@ -225,3 +219,171 @@ def get_publish_options(config): field = options.addChild('field', attrs={'var': var}) field.setTagData('value', value) return options + + +def _get_pubsub_items(response, node): + pubsub_node = response.getTag('pubsub', namespace=Namespace.PUBSUB) + if pubsub_node is None: + raise MalformedStanzaError('pubsub node missing', response) + + items_node = pubsub_node.getTag('items') + if items_node is None: + raise MalformedStanzaError('items node missing', response) + + if items_node.getAttr('node') != node: + raise MalformedStanzaError('invalid node attr', response) + + return items_node.getTags('item') + + +def _get_pubsub_item(response, node, id_): + items = _get_pubsub_items(response, node) + + if len(items) > 1: + raise MalformedStanzaError('multiple items found', response) + + if not items: + return None + + item = items[0] + if item.getAttr('id') != id_: + raise MalformedStanzaError('invalid item id', response) + + return item + + +def _make_pubsub_request(node, id_=None, max_items=None, jid=None): + query = Iq('get', to=jid) + pubsub = query.addChild('pubsub', namespace=Namespace.PUBSUB) + items = pubsub.addChild('items', {'node': node}) + if max_items is not None: + items.setAttr('max_items', max_items) + if id_ is not None: + items.addChild('item', {'id': id_}) + return query + + +def _get_configure_form(response, node): + pubsub = response.getTag('pubsub', namespace=Namespace.PUBSUB_OWNER) + if pubsub is None: + raise MalformedStanzaError('pubsub node missing', response) + + configure = pubsub.getTag('configure') + if configure is None: + raise MalformedStanzaError('configure node missing', response) + + if node != configure.getAttr('node'): + raise MalformedStanzaError('invalid node attribute', response) + + forms = configure.getTags('x', namespace=Namespace.DATA) + for form in forms: + dataform = extend_form(node=form) + form_type = dataform.vars.get('FORM_TYPE') + if form_type is None or form_type.value != Namespace.PUBSUB_CONFIG: + continue + + return dataform + + raise MalformedStanzaError('no valid form type found', response) + + +def _get_published_item_id(response, node, id_): + pubsub = response.getTag('pubsub', namespace=Namespace.PUBSUB) + if pubsub is None: + # https://xmpp.org/extensions/xep-0060.html#publisher-publish-success + # If the publish request did not include an ItemID, + # the IQ-result SHOULD include an empty <item/> element + # that specifies the ItemID of the published item. + # + # If the server did not add a payload we assume the item was + # published with the id we requested + return id_ + + publish = pubsub.getTag('publish') + if publish is None: + raise MalformedStanzaError('publish node missing', response) + + if node != publish.getAttr('node'): + raise MalformedStanzaError('invalid node attribute', response) + + item = publish.getTag('item') + if item is None: + raise MalformedStanzaError('item node missing', response) + + item_id = item.getAttr('id') + if id_ is not None and item_id != id_: + raise MalformedStanzaError('invalid item id', response) + + return item_id + + +def _make_publish_request(node, item, id_, options, jid): + query = Iq('set', to=jid) + pubsub = query.addChild('pubsub', namespace=Namespace.PUBSUB) + publish = pubsub.addChild('publish', {'node': node}) + attrs = {} + if id_ is not None: + attrs = {'id': id_} + publish.addChild('item', attrs, [item]) + if options: + publish = pubsub.addChild('publish-options') + publish.addChild(node=_make_publish_options(options)) + return query + + +def _make_publish_options(options): + data = Node(Namespace.DATA + ' x', attrs={'type': 'submit'}) + field = data.addChild('field', attrs={'var': 'FORM_TYPE', 'type': 'hidden'}) + field.setTagData('value', Namespace.PUBSUB_PUBLISH_OPTIONS) + + for var, value in options.items(): + field = data.addChild('field', attrs={'var': var}) + field.setTagData('value', value) + return data + + +def _make_retract_request(node, id_, jid, notify): + query = Iq('set', to=jid) + pubsub = query.addChild('pubsub', namespace=Namespace.PUBSUB) + attrs = {'node': node} + if notify: + attrs['notify'] = 'true' + retract = pubsub.addChild('retract', attrs=attrs) + retract.addChild('item', {'id': id_}) + return query + + +def _make_node_configuration(node, form, jid): + query = Iq('set', to=jid) + pubsub = query.addChild('pubsub', namespace=Namespace.PUBSUB_OWNER) + configure = pubsub.addChild('configure', {'node': node}) + form.setAttr('type', 'submit') + configure.addChild(node=form) + return query + + +def _make_node_configuration_request(node, jid): + query = Iq('get', to=jid) + pubsub = query.addChild('pubsub', namespace=Namespace.PUBSUB_OWNER) + pubsub.addChild('configure', {'node': node}) + return query + + +def _apply_options(form, options): + for var, value in options.items(): + try: + field = form[var] + except KeyError: + pass + else: + field.value = value + + +PubSubNodeConfigurationResult = namedtuple('PubSubConfigResult', + 'jid node form') + +PubSubConfigResult = namedtuple('PubSubConfigResult', + 'jid node form') + +PubSubPublishResult = namedtuple('PubSubPublishResult', + 'jid node id')

===================================== nbxmpp/modules/tune.py ===================================== @@ -24,6 +24,11 @@ from nbxmpp.modules.base import BaseModule

class Tune(BaseModule): + + _depends = { + 'publish': 'PubSub' + } + def __init__(self, client): BaseModule.__init__(self, client)

@@ -70,6 +75,5 @@ class Tune(BaseModule): if value is not None: item.addChild(tag, payload=value)

- jid = self._client.get_bound_jid().bare - self._client.get_module('PubSub').publish( - jid, Namespace.TUNE, item, id_='current') + task = self.publish(Namespace.TUNE, item, id_='current') + task.start()

===================================== nbxmpp/modules/user_avatar.py ===================================== @@ -15,22 +15,37 @@ # You should have received a copy of the GNU General Public License # along with this program; If not, see http://www.gnu.org/licenses/.

-import base64 +from typing import List +from typing import Dict + +import hashlib +from dataclasses import dataclass +from dataclasses import asdict +from dataclasses import field

from nbxmpp.namespaces import Namespace from nbxmpp.protocol import NodeProcessed -from nbxmpp.protocol import isResultNode +from nbxmpp.protocol import Node from nbxmpp.structs import StanzaHandler -from nbxmpp.structs import AvatarMetaData -from nbxmpp.structs import AvatarData -from nbxmpp.util import call_on_response -from nbxmpp.util import callback -from nbxmpp.util import raise_error -from nbxmpp.modules.pubsub import get_pubsub_request +from nbxmpp.structs import CommonResult +from nbxmpp.util import b64encode +from nbxmpp.util import b64decode +from nbxmpp.errors import MalformedStanzaError +from nbxmpp.task import xmpp_task from nbxmpp.modules.base import BaseModule +from nbxmpp.modules.util import raise_if_error +from nbxmpp.modules.util import finalize

class UserAvatar(BaseModule): + + _depends = { + 'get_node_configuration': 'PubSub', + 'set_node_configuration': 'PubSub', + 'publish': 'PubSub', + 'request_item': 'PubSub', + } + def __init__(self, client): BaseModule.__init__(self, client)

@@ -65,11 +80,10 @@ class UserAvatar(BaseModule): properties.jid) return

- info = metadata.getTags('info', one=True) try: - data = AvatarMetaData(**info.getAttrs()) - except Exception: - self._log.warning('Malformed user avatar data') + data = AvatarMetaData.from_node(metadata) + except Exception as error: + self._log.warning('Malformed user avatar data: %s', error) self._log.warning(stanza) raise NodeProcessed

@@ -79,42 +93,255 @@ class UserAvatar(BaseModule):

properties.pubsub_event = pubsub_event

- @call_on_response('_avatar_data_received') - def request_avatar(self, jid, id_): - return get_pubsub_request(jid, Namespace.AVATAR_DATA, id_=id_) + @xmpp_task + def _request_avatar(self, avatar_info, jid=None): + task = yield

- @callback - def _avatar_data_received(self, stanza): - jid = stanza.getFrom() - if jid is None: - jid = self._client.get_bound_jid().bare + item = yield self.request_item(Namespace.AVATAR_DATA, + id_=avatar_info.id, + jid=jid)

- if not isResultNode(stanza): - return raise_error(self._log.warning, stanza) + raise_if_error(item)

- pubsub_node = stanza.getTag('pubsub') - items_node = pubsub_node.getTag('items') - item = items_node.getTag('item') if item is None: - return raise_error(self._log.warning, stanza, 'stanza-malformed', - 'No item in node found') + yield task.set_result(None) + + yield _get_avatar_data(item, avatar_info.id) + + @xmpp_task + def set_avatar(self, avatar, public=False): + + task = yield + + if avatar is None: + result = yield self._publish_avatar_metadata(None) + yield finalize(task, result) + + access_model = 'open' if public else 'presence' + + result = yield self._publish_avatar(avatar, access_model) + + yield finalize(task, result) + + @xmpp_task + def _publish_avatar(self, avatar, access_model): + task = yield + + options = { + 'pubsub#persist_items': 'true', + 'pubsub#access_model': access_model, + } + + for info, data in avatar.pubsub_avatar_info(): + item = _make_avatar_data_node(data) + self._log.info('Publish avatar data: %s, %s', info, access_model) + + result = yield self.publish(Namespace.AVATAR_DATA, + item, + id_=info.id, + options=options, + force_node_options=True) + + raise_if_error(result) + + result = yield self._publish_avatar_metadata(avatar.metadata)

- sha = item.getAttr('id') - data_node = item.getTag('data', namespace=Namespace.AVATAR_DATA) - if data_node is None: - return raise_error(self._log.warning, stanza, 'stanza-malformed', - 'No data node found') + yield finalize(task, result)

- data = data_node.getData() - if data is None: - return raise_error(self._log.warning, stanza, 'stanza-malformed', - 'Data node empty') + @xmpp_task + def _publish_avatar_metadata(self, metadata): + task = yield + + self._log.info('Publish avatar meta data: %s', metadata) + + result = yield self.publish(Namespace.AVATAR_METADATA, + metadata.to_node(), + id_='current') + + yield finalize(task, result) + + @xmpp_task + def get_access_model(self): + _task = yield + + self._log.info('Request access model') + + result = yield self.get_node_configuration(Namespace.AVATAR_DATA) + + raise_if_error(result) + + yield result.form['pubsub#access_model'].value + + @xmpp_task + def set_access_model(self, model): + task = yield + + if model not in ('open', 'presence'): + raise ValueError('Invalid access model') + + result = yield self.get_node_configuration(Namespace.AVATAR_DATA) + + raise_if_error(result)

try: - data = base64.b64decode(data.encode('utf-8')) - except Exception as error: - return raise_error(self._log.warning, stanza, - 'stanza-malformed', str(error)) + access_model = result.form['pubsub#access_model'].value + except Exception: + yield task.set_error('warning', + condition='access-model-not-supported') + + if access_model == model: + jid = self._client.get_bound_jid().setBare() + yield CommonResult(jid=jid) + + result.form['pubsub#access_model'].value = model + + self._log.info('Set access model %s', model) + + result = yield self.set_node_configuration(Namespace.AVATAR_DATA, + result.form) + + yield finalize(task, result) + + +def _get_avatar_data(item, id_): + data_node = item.getTag('data', namespace=Namespace.AVATAR_DATA) + if data_node is None: + raise MalformedStanzaError('data node missing', item) + + data = data_node.getData() + if not data: + raise MalformedStanzaError('data node empty', item) + + try: + avatar = b64decode(data, return_type=bytes) + except Exception as error: + raise MalformedStanzaError(f'decoding error: {error}', item) + + avatar_sha = hashlib.sha1(avatar).hexdigest() + if avatar_sha != id_: + raise MalformedStanzaError(f'avatar does not match sha', item) + + return AvatarData(data=avatar, sha=avatar_sha) + + +def _make_metadata_node(infos): + item = Node('metadata', attrs={'xmlns': Namespace.AVATAR_METADATA}) + for info in infos: + item.addChild('info', attrs=info.to_dict()) + return item + + +def _make_avatar_data_node(avatar): + item = Node('data', attrs={'xmlns': Namespace.AVATAR_DATA}) + item.setData(b64encode(avatar.data)) + return item + + +def _get_info_attrs(avatar, avatar_sha, height, width): + info_attrs = { + 'id': avatar_sha, + 'bytes': len(avatar), + 'type': 'image/png', + } + + if height is not None: + info_attrs['height'] = height + + if width is not None: + info_attrs['width'] = width + + return info_attrs + + +@dataclass +class AvatarInfo: + bytes: int + id: str + type: str + url: str = None + height: int = None + width: int = None + + def __post_init__(self): + if self.bytes is None: + raise ValueError + if self.id is None: + raise ValueError + if self.type is None: + raise ValueError + + self.bytes = int(self.bytes) + + if self.height is not None: + self.height = int(self.height) + if self.width is not None: + self.width = int(self.width) + + def to_dict(self): + info_dict = asdict(self) + if self.height is None: + info_dict.pop('height') + if self.width is None: + info_dict.pop('width') + if self.url is None: + info_dict.pop('url') + return info_dict + + +@dataclass +class AvatarData: + data: bytes + sha: str + + +@dataclass +class AvatarMetaData: + infos: List[AvatarInfo] = field(default_factory=list) + + @classmethod + def from_node(cls, node): + infos = [] + info_nodes = node.getTags('info') + for info in info_nodes: + infos.append(AvatarInfo( + bytes=info.getAttr('bytes'), + id=info.getAttr('id'), + type=info.getAttr('type'), + url=info.getAttr('url'), + height=info.getAttr('height'), + width=info.getAttr('width') + )) + return cls(infos=infos) + + def add_avatar_info(self, avatar_info): + self.infos.append(avatar_info) + + def to_node(self): + return _make_metadata_node(self.infos) + + @property + def avatar_shas(self): + return [info.id for info in self.infos] + + +@dataclass +class Avatar: + metadata: AvatarMetaData = field(default_factory=AvatarMetaData) + data: Dict[AvatarInfo, bytes] = field(init=False, default_factory=dict) + + def add_image_source(self, data, type_, height, width, url=None): + sha = hashlib.sha1(data).hexdigest() + info = AvatarInfo(bytes=len(data), + id=sha, + type=type_, + height=height, + width=width, + url=url) + self.metadata.add_avatar_info(info) + self.data[info] = AvatarData(data=data, sha=sha)

- self._log.info('Received avatar data: %s %s', jid, sha) - return AvatarData(jid, sha, data) + def pubsub_avatar_info(self): + for info, data in self.data.items(): + if info.url is not None: + continue + yield info, data

===================================== nbxmpp/modules/util.py ===================================== @@ -0,0 +1,41 @@ +# Copyright (C) 2020 Philipp Hörist <philipp AT hoerist.com> +# +# This file is part of nbxmpp. +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; either version 3 +# of the License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; If not, see http://www.gnu.org/licenses/. + +from nbxmpp.structs import CommonResult +from nbxmpp.errors import StanzaError +from nbxmpp.errors import is_error +from nbxmpp.simplexml import Node + + +def process_response(response): + if response.isError(): + raise StanzaError(response) + + return CommonResult(jid=response.getFrom()) + + +def raise_if_error(result): + if is_error(result): + raise result + + +def finalize(task, result): + if is_error(result): + raise result + if isinstance(result, Node): + return task.set_result(result) + return result

===================================== nbxmpp/modules/vcard_temp.py ===================================== @@ -0,0 +1,143 @@ +# Copyright (C) 2020 Philipp Hörist <philipp AT hoerist.com> +# +# This file is part of nbxmpp. +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; either version 3 +# of the License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; If not, see http://www.gnu.org/licenses/. + +import hashlib +from dataclasses import dataclass +from dataclasses import field + +from nbxmpp.task import xmpp_task +from nbxmpp.util import b64decode +from nbxmpp.util import b64encode +from nbxmpp.errors import StanzaError +from nbxmpp.errors import MalformedStanzaError +from nbxmpp.protocol import Iq +from nbxmpp.simplexml import Node +from nbxmpp.namespaces import Namespace +from nbxmpp.modules.base import BaseModule +from nbxmpp.modules.util import process_response + + +class VCardTemp(BaseModule): + def __init__(self, client): + BaseModule.__init__(self, client) + + self._client = client + self.handlers = [] + + @xmpp_task + def request_vcard(self, jid=None): + _task = yield + + response = yield _make_vcard_request(jid) + + if response.isError(): + raise StanzaError(response) + + vcard_node = _get_vcard_node(response) + yield VCard.from_node(vcard_node) + + @xmpp_task + def set_vcard(self, vcard, jid=None): + _task = yield + + response = yield _make_vcard_publish(jid, vcard) + yield process_response(response) + + +def _make_vcard_request(jid): + iq = Iq(typ='get', to=jid) + iq.addChild('vCard', namespace=Namespace.VCARD) + return iq + + +def _get_vcard_node(response): + vcard_node = response.getTag('vCard', namespace=Namespace.VCARD) + if vcard_node is None: + raise MalformedStanzaError('vCard node missing', response) + return vcard_node + + +def _make_vcard_publish(jid, vcard): + iq = Iq(typ='set', to=jid) + iq.addChild(node=vcard.to_node()) + return iq + + +@dataclass +class VCard: + data: dict = field(default_factory=dict) + + @classmethod + def from_node(cls, node): + dict_ = {} + for info in node.getChildren(): + name = info.getName() + if name in ('ADR', 'TEL', 'EMAIL'): + dict_.setdefault(name, []) + entry = {} + for child in info.getChildren(): + entry[child.getName()] = child.getData() + dict_[name].append(entry) + elif info.getChildren() == []: + dict_[name] = info.getData() + else: + dict_[name] = {} + for child in info.getChildren(): + dict_[name][child.getName()] = child.getData() + + return cls(data=dict_) + + def to_node(self): + vcard = Node(tag='vCard', attrs={'xmlns': Namespace.VCARD}) + for i in self.data: + if i == 'jid': + continue + if isinstance(self.data[i], dict): + child = vcard.addChild(i) + for j in self.data[i]: + child.addChild(j).setData(self.data[i][j]) + elif isinstance(self.data[i], list): + for j in self.data[i]: + child = vcard.addChild(i) + for k in j: + child.addChild(k).setData(j[k]) + else: + vcard.addChild(i).setData(self.data[i]) + return vcard + + def set_avatar(self, avatar, type_=None): + avatar = b64encode(avatar) + if 'PHOTO' not in self.data: + self.data['PHOTO'] = {} + + self.data['PHOTO']['BINVAL'] = avatar + + if type_ is not None: + self.data['PHOTO']['TYPE'] = type_ + + def get_avatar(self): + try: + avatar = self.data['PHOTO']['BINVAL'] + except Exception: + return None, None + + if not avatar: + return None, None + + avatar = b64decode(avatar, return_type=bytes) + avatar_sha = hashlib.sha1(avatar).hexdigest() + return avatar, avatar_sha

===================================== nbxmpp/namespaces.py ===================================== @@ -121,6 +121,7 @@ class _Namespaces: PUBKEY_PUBKEY: str = 'urn:xmpp:pubkey:2' PUBKEY_REVOKE: str = 'urn:xmpp:revoke:2' PUBSUB: str = 'http://jabber.org/protocol/pubsub' + PUBSUB_ERROR: str = 'http://jabber.org/protocol/pubsub#errors' PUBSUB_CONFIG: str = 'http://jabber.org/protocol/pubsub#node_config' PUBSUB_EVENT: str = 'http://jabber.org/protocol/pubsub#event' PUBSUB_OWNER: str = 'http://jabber.org/protocol/pubsub#owner'

===================================== nbxmpp/protocol.py ===================================== @@ -1072,6 +1072,12 @@ class Protocol(Node): if timestamp is not None: self.setTimestamp(timestamp)

+ def isError(self): + return self.getAttr('type') == 'error' + + def isResult(self): + return self.getAttr('type') == 'result' + def getTo(self): """ Return value of the 'to' attribute

===================================== nbxmpp/structs.py ===================================== @@ -32,6 +32,7 @@ from nbxmpp.const import PresenceType from nbxmpp.const import LOCATION_DATA from nbxmpp.const import AdHocStatus

+ StanzaHandler = namedtuple('StanzaHandler', 'name callback typ ns xmlns priority') StanzaHandler.__new__.__defaults__ = ('', '', None, 50) @@ -73,10 +74,6 @@ StanzaIDData.__new__.__defaults__ = (None, None) PubSubEventData = namedtuple('PubSubEventData', 'node id item data deleted retracted purged') PubSubEventData.__new__.__defaults__ = (None, None, None, False, False, False)

-PubSubConfigResult = namedtuple('PubSubConfigResult', 'jid node form') - -PubSubPublishResult = namedtuple('PubSubPublishResult', 'jid node id') - MoodData = namedtuple('MoodData', 'mood text')

ActivityData = namedtuple('ActivityData', 'activity subactivity text') @@ -84,12 +81,6 @@ ActivityData = namedtuple('ActivityData', 'activity subactivity text') LocationData = namedtuple('LocationData', LOCATION_DATA) LocationData.__new__.__defaults__ = (None,) * len(LocationData._fields)

-AvatarMetaData = namedtuple('AvatarMetaData', 'bytes height width id type url') -AvatarMetaData.__new__.__defaults__ = (None,) * len(AvatarMetaData._fields) - -AvatarData = namedtuple('AvatarData', 'jid sha data') -AvatarData.__new__.__defaults__ = (None,) * len(AvatarData._fields) - BookmarkData = namedtuple('BookmarkData', 'jid name nick autojoin password') BookmarkData.__new__.__defaults__ = (None, None, None, None)

===================================== nbxmpp/task.py ===================================== @@ -0,0 +1,257 @@ +# Copyright (C) 2020 Philipp Hörist <philipp AT hoerist.com> +# +# This file is part of nbxmpp. +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; either version 3 +# of the License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; If not, see http://www.gnu.org/licenses/. + +import weakref +import inspect +import logging +from enum import IntEnum +from functools import wraps + +from nbxmpp.errors import is_error +from nbxmpp.simplexml import Node + + +log = logging.getLogger('nbxmpp.task') + + +class _ResultSet: + pass + +ResultSet = _ResultSet() + + +class NoType: + pass + + +class TaskState(IntEnum): + INIT = 0 + RUNNING = 1 + FINISHED = 2 + CANCELLED = 3 + + @property + def is_init(self): + return self == TaskState.INIT + + @property + def is_running(self): + return self == TaskState.RUNNING + + @property + def is_finished(self): + return self == TaskState.FINISHED + + @property + def is_cancelled(self): + return self == TaskState.CANCELLED + + +def xmpp_task(func): + @wraps(func) + def func_wrapper(self, *args, + callback=None, user_data=None, **kwargs): + task = XMPPTask(func(self, *args, **kwargs), self._log, self._client) + self._client.add_task(task) + task.set_finalize_func(self._client.remove_task) + task.set_user_data(user_data) + if callback is not None: + task.add_done_callback(callback) + task.start() + return task + return func_wrapper + + +def is_fatal_error(error): + if is_error(error): + return error.is_fatal + return isinstance(error, Exception) + + +class Task: + + _process_types = (NoType,) + + def __init__(self, gen, logger=log): + self._logger = logger + self._gen = gen + self._done_callbacks = [] + self._sub_task = None + self._result = None + self._error = None + self._user_data = None + self._finalize_func = None + self._finalize_context = None + self._state = TaskState.INIT + + @property + def state(self): + return self._state + + def add_done_callback(self, callback, weak=True): + if self._state in (TaskState.FINISHED, TaskState.CANCELLED): + raise RuntimeError('Task is finished') + + if weak: + if inspect.ismethod(callback): + callback = weakref.WeakMethod(callback) + elif inspect.isfunction(callback): + callback = weakref.ref(callback) + else: + ValueError('Unknown callback object') + + self._done_callbacks.append(callback) + + def start(self): + if self._state != TaskState.INIT: + raise RuntimeError('Task already started') + + self._state = TaskState.RUNNING + next(self._gen) + self._next_step(self) + + def _run_async(self, data): + raise NotImplementedError + + def _async_finished(self, *args, **kwargs): + raise NotImplementedError + + def _sub_task_completed(self, task): + self._sub_task = None + if self._state != TaskState.CANCELLED: + result = task.get_result() + if is_fatal_error(result): + self._error = result + self._set_finished() + else: + self._next_step(result) + + def _next_step(self, result): + try: + res = self._gen.send(result) + except StopIteration: + self._set_finished() + return + + except Exception as error: + self._log_if_fatal(error) + self._error = error + self._set_finished() + return + + if isinstance(res, self._process_types): + self._run_async(res) + + elif isinstance(res, Task): + if self._sub_task is not None: + RuntimeError('Only one sub task can be active') + + self._sub_task = res + self._sub_task.add_done_callback(self._sub_task_completed, + weak=False) + self._sub_task.start() + + else: + if res is not ResultSet: + self._result = res + self._set_finished() + + def _set_finished(self): + self._state = TaskState.FINISHED + self._invoke_callbacks() + self._finalize() + + def _log_if_fatal(self, error): + if is_error(error): + if error.is_fatal: + self._logger.log(error.log_level, error) + + elif isinstance(error, Exception): + self._logger.exception('Fatal Exception') + + def _invoke_callbacks(self): + for callback in self._done_callbacks: + if isinstance(callback, weakref.WeakMethod): + callback = callback() + if callback is None: + return + try: + callback(self) + except Exception: + self._logger.exception('Exception while executing callback') + + def set_result(self, result): + self._result = result + return ResultSet + + def get_result(self): + # if self._error is None, there was no error + # but None is a valid value for self._result + if self._error is not None: + return self._error + return self._result + + def finish(self): + if self._error is not None: + raise self._error + return self._result + + def set_user_data(self, user_data): + self._user_data = user_data + + def get_user_data(self): + return self._user_data + + def set_finalize_func(self, func, context=None): + self._finalize_func = func + self._finalize_context = context + + def cancel(self): + self._state = TaskState.CANCELLED + if self._sub_task is not None: + self._sub_task.cancel() + self._finalize() + + def _finalize(self): + self._done_callbacks.clear() + self._sub_task = None + self._error = None + self._result = None + self._user_data = None + self._gen.close() + if self._finalize_func is not None: + self._finalize_func(self, self._finalize_context) + + +class XMPPTask(Task): + + _process_types = (Node,) + + def __init__(self, gen, logger, client): + super().__init__(gen, logger) + self._client = client + + def _run_async(self, stanza): + self._client.send_stanza(stanza, callback=self._async_finished) + + def _async_finished(self, _client, result, *args, **kwargs): + if self._state != TaskState.CANCELLED: + self._next_step(result) + + def _finalize(self): + self._client = None + super()._finalize()

===================================== python-nbxmpp.doap ===================================== @@ -158,6 +158,13 @@ xmpp:version1.1</xmpp:version> </xmpp:SupportedXep> </implements> + <implements> + xmpp:SupportedXep + <xmpp:xep rdf:resource="https://xmpp.org/extensions/xep-0106.html" /> + xmpp:statuscomplete</xmpp:status> + xmpp:version1.1.1</xmpp:version> + </xmpp:SupportedXep> + </implements> <implements> xmpp:SupportedXep <xmpp:xep rdf:resource="https://xmpp.org/extensions/xep-0107.html" />

View it on GitLab: https://dev.gajim.org/gajim/python-nbxmpp/-/compare/cecc58aeabf55c7f04b39227...