libervia-backend: sat/plugins/plugin_xep

comparison sat/plugins/plugin_xep_0060.py @ 3213:c2f958dde5d2

plugin XEP-0060: async sendItems + precondition-not-met policy: - sendItems is now an "async" coroutine, and sendItem use it instead of duplicating publication logic - policy to use when a precondition is not met (with publish-options) can now be specified, for now it's either raise the exception (default policy), or try to publish without the options - constants have been added to handle "extra" keys

author	Goffi <goffi@goffi.org>
date	Wed, 11 Mar 2020 19:15:48 +0100
parents	559a625a236b
children	4c98f4972db5

comparison

equal deleted inserted replaced

-:89d97776fd34
+:c2f958dde5d2
 #!/usr/bin/env python3
+# SàT plugin for Publish-Subscribe (xep-0060)
-# SAT plugin for Publish-Subscribe (xep-0060)
 # Copyright (C) 2009-2020 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
 # the Free Software Foundation, either version 3 of the License, or
 from collections import namedtuple
 import urllib.request, urllib.parse, urllib.error
 from functools import reduce
-from sat.core.i18n import _
-from sat.core.constants import Const as C
-from sat.core.log import getLogger
-from sat.core import exceptions
-from sat.tools import sat_defer
-from sat.tools.common import data_format
 from zope.interface import implementer
+from twisted.words.xish import domish
+from twisted.words.protocols.jabber import jid, error
+from twisted.internet import reactor, defer
 from wokkel import disco
 from wokkel import data_form
 from wokkel import generic
 # XXX: sat_tmp.wokkel.pubsub is actually use instead of wokkel version
 #      mam and rsm come from sat_tmp.wokkel too
 from wokkel import pubsub
 from wokkel import rsm
 from wokkel import mam
+from sat.core.i18n import _
-from twisted.words.protocols.jabber import jid, error
+from sat.core.constants import Const as C
-from twisted.internet import reactor, defer
+from sat.core.log import getLogger
+from sat.core import exceptions
+from sat.tools import sat_defer
+from sat.tools.common import data_format
 log = getLogger(__name__)
 PLUGIN_INFO = {
 ACCESS_ROSTER = "roster"
 ACCESS_PUBLISHER_ROSTER = "publisher-roster"
 ACCESS_AUTHORIZE = "authorize"
 ACCESS_WHITELIST = "whitelist"
 ID_SINGLETON = "current"
+EXTRA_PUBLISH_OPTIONS = "publish_options"
+EXTRA_ON_PRECOND_NOT_MET = "on_precondition_not_met"
 def __init__(self, host):
 log.info(_("PubSub plugin initialization"))
 self.host = host
 self._rsm = host.plugins.get("XEP-0059")
 d = self.sendItems(
 client, service, nodeIdentifier, items, extra
 )
 return d
-def _getPublishedItemId(self, iq_elt, original_id):
+def _getPublishedItemId(self, published_ids, original_id):
-"""return item of published id if found in answer
+"""Return item of published id if found in answer
-if not found original_id is returned, or empty string if it is None or empty
+if not found original_id is returned, which may be None
-string
 """
 try:
-item_id = iq_elt.pubsub.publish.item["id"]
+return published_ids[0]
-except (AttributeError, KeyError):
+except IndexError:
-item_id = None
+return original_id
-return item_id or original_id
 def sendItem(self, client, service, nodeIdentifier, payload, item_id=None,
 extra=None):
 """High level method to send one item
 @param payload(domish.Element, unicode): payload of the item to send
 @param item_id(unicode, None): id to use or None to create one
 @param extra(dict, None): extra option, not used yet
 @return (unicode, None): id of the created item
 """
-if extra is None:
+item_elt = domish.Element((pubsub.NS_PUBSUB, 'item'))
-extra = {}
+if item_id is not None:
-publish_options = extra.get('publish_options')
+item_elt['id'] = item_id
-item_elt = pubsub.Item(id=item_id, payload=payload)
+item_elt.addChild(payload)
-d = self.publish(
+d = defer.ensureDeferred(self.sendItems(
-client, service, nodeIdentifier, [item_elt], options=publish_options)
+client,
+service,
+nodeIdentifier,
+[item_elt],
+extra
+))
 d.addCallback(self._getPublishedItemId, item_id)
 return d
-def _publishCb(self, iq_result):
+async def sendItems(self, client, service, nodeIdentifier, items, extra=None):
-"""Parse publish result, and return ids given by pubsub service"""
-try:
-item_ids = [item['id']
-for item in iq_result.pubsub.publish.elements(pubsub.NS_PUBSUB, 'item')]
-except AttributeError:
-return []
-return item_ids
-def sendItems(self, client, service, nodeIdentifier, items, extra=None):
 """High level method to send several items at once
 @param service(jid.JID, None): service to send the item to
 None to use PEP
 @param NodeIdentifier(unicode): PubSub node to use
 @param items(list[domish.Element]): whole item elements to send,
 "id" will be used if set
 @param extra(dict, None): extra options. Key can be:
-- publish_options: dict of publish-options
+- self.EXTRA_PUBLISH_OPTIONS(dict): publish options, cf. XEP-0060 § 7.1.5
+the dict maps option name to value(s)
+- self.EXTRA_ON_PRECOND_NOT_MET(str): policy to have when publishing is
+failing du to failing precondition. Value can be:
+* raise (default): raise the exception
+* publish_without_options: re-publish without the publish-options.
+A warning will be logged showing that the publish-options could not
+be used
 @return (list[unicode]): ids of the created items
 """
 if extra is None:
 extra = {}
 parsed_items = []
 for item in items:
 if item.name != 'item':
 raise exceptions.DataError(_("Invalid item: {xml}").format(item.toXml()))
 item_id = item.getAttribute("id")
 parsed_items.append(pubsub.Item(id=item_id, payload=item.firstChildElement()))
-publish_options = extra.get('publish_options')
+publish_options = extra.get(self.EXTRA_PUBLISH_OPTIONS)
-d = self.publish(
+try:
-client, service, nodeIdentifier, parsed_items, options=publish_options)
+iq_result = await self.publish(
-d.addCallback(self._publishCb)
+client, service, nodeIdentifier, parsed_items, options=publish_options)
-return d
+except error.StanzaError as e:
+if ((e.condition == 'conflict' and e.appCondition
+and e.appCondition.name == 'precondition-not-met'
+and publish_options is not None)):
+# this usually happens when publish-options can't be set
+policy = extra.get(self.EXTRA_ON_PRECOND_NOT_MET, 'raise')
+if policy == 'raise':
+raise e
+elif policy == 'publish_without_options':
+log.warning(_(
+"Can't use publish-options ({options}) on node {node}, "
+"re-publishing without them: {reason}").format(
+options=', '.join(f'{k} = {v}'
+for k,v in publish_options.items()),
+node=nodeIdentifier,
+reason=e,
+)
+)
+iq_result = await self.publish(
+client, service, nodeIdentifier, parsed_items)
+else:
+raise exceptions.InternalError(
+f"Invalid policy in extra's {self.EXTRA_ON_PRECOND_NOT_MET!r}: "
+f"{policy}"
+)
+else:
+raise e
+try:
+return [item['id']
+for item in iq_result.pubsub.publish.elements(pubsub.NS_PUBSUB, 'item')]
+except AttributeError:
+return []
 def publish(self, client, service, nodeIdentifier, items=None, options=None):
 return client.pubsub_client.publish(
 service, nodeIdentifier, items, client.pubsub_client.parent.jid,
 options=options

Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0060.py @ 3213:c2f958dde5d2