1652 lines
54 KiB
Python
Executable file
1652 lines
54 KiB
Python
Executable file
# -*- coding: utf-8 -*-
|
|
# -*- test-case-name: wokkel.test.test_pubsub -*-
|
|
#
|
|
# SàT adaptation for wokkel.pubsub
|
|
# Copyright (C) 2015 Adien Cossa (souliane@mailoo.org)
|
|
# Copyright (c) 2003-2012 Ralph Meijer.
|
|
|
|
# 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
|
|
# (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 Affero General Public License for more details.
|
|
|
|
# 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/>.
|
|
# --
|
|
|
|
# This program is based on wokkel (https://wokkel.ik.nu/),
|
|
# originaly written by Ralph Meijer
|
|
# It is sublicensed under AGPL v3 (or any later version) as allowed by the original
|
|
# license.
|
|
|
|
# --
|
|
|
|
# Here is a copy of the original license:
|
|
|
|
# Copyright (c) 2003-2012 Ralph Meijer.
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining
|
|
# a copy of this software and associated documentation files (the
|
|
# "Software"), to deal in the Software without restriction, including
|
|
# without limitation the rights to use, copy, modify, merge, publish,
|
|
# distribute, sublicense, and/or sell copies of the Software, and to
|
|
# permit persons to whom the Software is furnished to do so, subject to
|
|
# the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be
|
|
# included in all copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
|
|
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
|
|
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
|
|
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
|
|
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
|
|
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
|
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
|
|
|
"""
|
|
XMPP publish-subscribe protocol.
|
|
|
|
This protocol is specified in
|
|
U{XEP-0060<http://xmpp.org/extensions/xep-0060.html>}.
|
|
"""
|
|
|
|
from zope.interface import implements
|
|
|
|
from twisted.internet import defer
|
|
from twisted.python import log
|
|
from twisted.words.protocols.jabber import jid, error
|
|
from twisted.words.xish import domish
|
|
|
|
from wokkel import disco, data_form, generic, shim
|
|
from wokkel.compat import IQ
|
|
from wokkel.subprotocols import IQHandlerMixin, XMPPHandler
|
|
from wokkel.iwokkel import IPubSubClient, IPubSubService, IPubSubResource
|
|
|
|
# Iq get and set XPath queries
|
|
IQ_GET = '/iq[@type="get"]'
|
|
IQ_SET = '/iq[@type="set"]'
|
|
|
|
# Publish-subscribe namespaces
|
|
NS_PUBSUB = 'http://jabber.org/protocol/pubsub'
|
|
NS_PUBSUB_EVENT = NS_PUBSUB + '#event'
|
|
NS_PUBSUB_ERRORS = NS_PUBSUB + '#errors'
|
|
NS_PUBSUB_OWNER = NS_PUBSUB + "#owner"
|
|
NS_PUBSUB_NODE_CONFIG = NS_PUBSUB + "#node_config"
|
|
NS_PUBSUB_META_DATA = NS_PUBSUB + "#meta-data"
|
|
NS_PUBSUB_SUBSCRIBE_OPTIONS = NS_PUBSUB + "#subscribe_options"
|
|
|
|
# XPath to match pubsub requests
|
|
PUBSUB_REQUEST = '/iq[@type="get" or @type="set"]/' + \
|
|
'pubsub[@xmlns="' + NS_PUBSUB + '" or ' + \
|
|
'@xmlns="' + NS_PUBSUB_OWNER + '"]'
|
|
|
|
BOOL_TRUE = ('1','true')
|
|
BOOL_FALSE = ('0','false')
|
|
|
|
class SubscriptionPending(Exception):
|
|
"""
|
|
Raised when the requested subscription is pending acceptance.
|
|
"""
|
|
|
|
|
|
|
|
class SubscriptionUnconfigured(Exception):
|
|
"""
|
|
Raised when the requested subscription needs to be configured before
|
|
becoming active.
|
|
"""
|
|
|
|
|
|
|
|
class PubSubError(error.StanzaError):
|
|
"""
|
|
Exception with publish-subscribe specific condition.
|
|
"""
|
|
def __init__(self, condition, pubsubCondition, feature=None, text=None):
|
|
appCondition = domish.Element((NS_PUBSUB_ERRORS, pubsubCondition))
|
|
if feature:
|
|
appCondition['feature'] = feature
|
|
error.StanzaError.__init__(self, condition,
|
|
text=text,
|
|
appCondition=appCondition)
|
|
|
|
|
|
|
|
class BadRequest(error.StanzaError):
|
|
"""
|
|
Bad request stanza error.
|
|
"""
|
|
def __init__(self, pubsubCondition=None, text=None):
|
|
if pubsubCondition:
|
|
appCondition = domish.Element((NS_PUBSUB_ERRORS, pubsubCondition))
|
|
else:
|
|
appCondition = None
|
|
error.StanzaError.__init__(self, 'bad-request',
|
|
text=text,
|
|
appCondition=appCondition)
|
|
|
|
|
|
|
|
class Unsupported(PubSubError):
|
|
def __init__(self, feature, text=None):
|
|
self.feature = feature
|
|
PubSubError.__init__(self, 'feature-not-implemented',
|
|
'unsupported',
|
|
feature,
|
|
text)
|
|
|
|
def __str__(self):
|
|
message = PubSubError.__str__(self)
|
|
message += ', feature %r' % self.feature
|
|
return message
|
|
|
|
|
|
class Subscription(object):
|
|
"""
|
|
A subscription to a node.
|
|
|
|
@ivar nodeIdentifier: The identifier of the node subscribed to. The root
|
|
node is denoted by C{None}.
|
|
@type nodeIdentifier: C{unicode}
|
|
|
|
@ivar subscriber: The subscribing entity.
|
|
@type subscriber: L{jid.JID}
|
|
|
|
@ivar state: The subscription state. One of C{'subscribed'}, C{'pending'},
|
|
C{'unconfigured'}.
|
|
@type state: C{unicode}
|
|
|
|
@ivar options: Optional list of subscription options.
|
|
@type options: C{dict}
|
|
|
|
@ivar subscriptionIdentifier: Optional subscription identifier.
|
|
@type subscriptionIdentifier: C{unicode}
|
|
"""
|
|
|
|
def __init__(self, nodeIdentifier, subscriber, state, options=None,
|
|
subscriptionIdentifier=None):
|
|
self.nodeIdentifier = nodeIdentifier
|
|
self.subscriber = subscriber
|
|
self.state = state
|
|
self.options = options or {}
|
|
self.subscriptionIdentifier = subscriptionIdentifier
|
|
|
|
|
|
@staticmethod
|
|
def fromElement(element):
|
|
return Subscription(
|
|
element.getAttribute('node'),
|
|
jid.JID(element.getAttribute('jid')),
|
|
element.getAttribute('subscription'),
|
|
subscriptionIdentifier=element.getAttribute('subid'))
|
|
|
|
|
|
def toElement(self, defaultUri=None):
|
|
"""
|
|
Return the DOM representation of this subscription.
|
|
|
|
@rtype: L{domish.Element}
|
|
"""
|
|
element = domish.Element((defaultUri, 'subscription'))
|
|
if self.nodeIdentifier:
|
|
element['node'] = self.nodeIdentifier
|
|
element['jid'] = unicode(self.subscriber)
|
|
element['subscription'] = self.state
|
|
if self.subscriptionIdentifier:
|
|
element['subid'] = self.subscriptionIdentifier
|
|
return element
|
|
|
|
|
|
|
|
class Item(domish.Element):
|
|
"""
|
|
Publish subscribe item.
|
|
|
|
This behaves like an object providing L{domish.IElement}.
|
|
|
|
Item payload can be added using C{addChild} or C{addRawXml}, or using the
|
|
C{payload} keyword argument to C{__init__}.
|
|
"""
|
|
|
|
def __init__(self, id=None, payload=None):
|
|
"""
|
|
@param id: optional item identifier
|
|
@type id: C{unicode}
|
|
@param payload: optional item payload. Either as a domish element, or
|
|
as serialized XML.
|
|
@type payload: object providing L{domish.IElement} or C{unicode}.
|
|
"""
|
|
|
|
domish.Element.__init__(self, (None, 'item'))
|
|
if id is not None:
|
|
self['id'] = id
|
|
if payload is not None:
|
|
if isinstance(payload, basestring):
|
|
self.addRawXml(payload)
|
|
else:
|
|
self.addChild(payload)
|
|
|
|
|
|
|
|
class PubSubRequest(generic.Stanza):
|
|
"""
|
|
A publish-subscribe request.
|
|
|
|
The set of instance variables used depends on the type of request. If
|
|
a variable is not applicable or not passed in the request, its value is
|
|
C{None}.
|
|
|
|
@ivar verb: The type of publish-subscribe request. See C{_requestVerbMap}.
|
|
@type verb: C{str}.
|
|
|
|
@ivar affiliations: Affiliations to be modified.
|
|
@type affiliations: C{set}
|
|
|
|
@ivar items: The items to be published, as L{domish.Element}s.
|
|
@type items: C{list}
|
|
|
|
@ivar itemIdentifiers: Identifiers of the items to be retrieved or
|
|
retracted.
|
|
@type itemIdentifiers: C{set}
|
|
|
|
@ivar maxItems: Maximum number of items to retrieve.
|
|
@type maxItems: C{int}.
|
|
|
|
@ivar nodeIdentifier: Identifier of the node the request is about.
|
|
@type nodeIdentifier: C{unicode}
|
|
|
|
@ivar nodeType: The type of node that should be created, or for which the
|
|
configuration is retrieved. C{'leaf'} or C{'collection'}.
|
|
@type nodeType: C{str}
|
|
|
|
@ivar options: Configurations options for nodes, subscriptions and publish
|
|
requests.
|
|
@type options: L{data_form.Form}
|
|
|
|
@ivar subscriber: The subscribing entity.
|
|
@type subscriber: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@ivar subscriptionIdentifier: Identifier for a specific subscription.
|
|
@type subscriptionIdentifier: C{unicode}
|
|
|
|
@ivar subscriptions: Subscriptions to be modified, as a set of
|
|
L{Subscription}.
|
|
@type subscriptions: C{set}
|
|
|
|
@ivar affiliations: Affiliations to be modified, as a dictionary of entity
|
|
(L{JID<twisted.words.protocols.jabber.jid.JID>} to affiliation
|
|
(C{unicode}).
|
|
@type affiliations: C{dict}
|
|
"""
|
|
|
|
verb = None
|
|
|
|
affiliations = None
|
|
items = None
|
|
itemIdentifiers = None
|
|
maxItems = None
|
|
nodeIdentifier = None
|
|
nodeType = None
|
|
options = None
|
|
subscriber = None
|
|
subscriptionIdentifier = None
|
|
subscriptions = None
|
|
affiliations = None
|
|
notify = None
|
|
|
|
# Map request iq type and subelement name to request verb
|
|
_requestVerbMap = {
|
|
('set', NS_PUBSUB, 'publish'): 'publish',
|
|
('set', NS_PUBSUB, 'subscribe'): 'subscribe',
|
|
('set', NS_PUBSUB, 'unsubscribe'): 'unsubscribe',
|
|
('get', NS_PUBSUB, 'options'): 'optionsGet',
|
|
('set', NS_PUBSUB, 'options'): 'optionsSet',
|
|
('get', NS_PUBSUB, 'subscriptions'): 'subscriptions',
|
|
('get', NS_PUBSUB, 'affiliations'): 'affiliations',
|
|
('set', NS_PUBSUB, 'create'): 'create',
|
|
('get', NS_PUBSUB_OWNER, 'default'): 'default',
|
|
('get', NS_PUBSUB_OWNER, 'configure'): 'configureGet',
|
|
('set', NS_PUBSUB_OWNER, 'configure'): 'configureSet',
|
|
('get', NS_PUBSUB, 'items'): 'items',
|
|
('set', NS_PUBSUB, 'retract'): 'retract',
|
|
('set', NS_PUBSUB_OWNER, 'purge'): 'purge',
|
|
('set', NS_PUBSUB_OWNER, 'delete'): 'delete',
|
|
('get', NS_PUBSUB_OWNER, 'affiliations'): 'affiliationsGet',
|
|
('set', NS_PUBSUB_OWNER, 'affiliations'): 'affiliationsSet',
|
|
('get', NS_PUBSUB_OWNER, 'subscriptions'): 'subscriptionsGet',
|
|
('set', NS_PUBSUB_OWNER, 'subscriptions'): 'subscriptionsSet',
|
|
}
|
|
|
|
# Map request verb to request iq type and subelement name
|
|
_verbRequestMap = dict(((v, k) for k, v in _requestVerbMap.iteritems()))
|
|
|
|
# Map request verb to parameter handler names
|
|
_parameters = {
|
|
'publish': ['node', 'items'],
|
|
'subscribe': ['nodeOrEmpty', 'jid', 'optionsWithSubscribe'],
|
|
'unsubscribe': ['nodeOrEmpty', 'jid', 'subidOrNone'],
|
|
'optionsGet': ['nodeOrEmpty', 'jid', 'subidOrNone'],
|
|
'optionsSet': ['nodeOrEmpty', 'jid', 'options', 'subidOrNone'],
|
|
'subscriptions': [],
|
|
'affiliations': [],
|
|
'create': ['nodeOrNone', 'configureOrNone'],
|
|
'default': ['default'],
|
|
'configureGet': ['nodeOrEmpty'],
|
|
'configureSet': ['nodeOrEmpty', 'configureOrNone'],
|
|
'items': ['node', 'maxItems', 'itemIdentifiers', 'subidOrNone'],
|
|
'retract': ['node', 'notify', 'itemIdentifiers'],
|
|
'purge': ['node'],
|
|
'delete': ['node'],
|
|
'affiliationsGet': ['nodeOrEmpty'],
|
|
'affiliationsSet': ['nodeOrEmpty', 'affiliations'],
|
|
'subscriptionsGet': ['nodeOrEmpty'],
|
|
'subscriptionsSet': [],
|
|
}
|
|
|
|
def __init__(self, verb=None):
|
|
self.verb = verb
|
|
|
|
|
|
def _parse_node(self, verbElement):
|
|
"""
|
|
Parse the required node identifier out of the verbElement.
|
|
"""
|
|
try:
|
|
self.nodeIdentifier = verbElement["node"]
|
|
except KeyError:
|
|
raise BadRequest('nodeid-required')
|
|
|
|
|
|
def _render_node(self, verbElement):
|
|
"""
|
|
Render the required node identifier on the verbElement.
|
|
"""
|
|
if not self.nodeIdentifier:
|
|
raise Exception("Node identifier is required")
|
|
|
|
verbElement['node'] = self.nodeIdentifier
|
|
|
|
|
|
def _parse_nodeOrEmpty(self, verbElement):
|
|
"""
|
|
Parse the node identifier out of the verbElement. May be empty.
|
|
"""
|
|
self.nodeIdentifier = verbElement.getAttribute("node", '')
|
|
|
|
|
|
def _render_nodeOrEmpty(self, verbElement):
|
|
"""
|
|
Render the node identifier on the verbElement. May be empty.
|
|
"""
|
|
if self.nodeIdentifier:
|
|
verbElement['node'] = self.nodeIdentifier
|
|
|
|
|
|
def _parse_nodeOrNone(self, verbElement):
|
|
"""
|
|
Parse the optional node identifier out of the verbElement.
|
|
"""
|
|
self.nodeIdentifier = verbElement.getAttribute("node")
|
|
|
|
|
|
def _render_nodeOrNone(self, verbElement):
|
|
"""
|
|
Render the optional node identifier on the verbElement.
|
|
"""
|
|
if self.nodeIdentifier:
|
|
verbElement['node'] = self.nodeIdentifier
|
|
|
|
|
|
def _parse_items(self, verbElement):
|
|
"""
|
|
Parse items out of the verbElement for publish requests.
|
|
"""
|
|
self.items = []
|
|
for element in verbElement.elements():
|
|
if element.uri == NS_PUBSUB and element.name == 'item':
|
|
self.items.append(element)
|
|
|
|
|
|
def _render_items(self, verbElement):
|
|
"""
|
|
Render items into the verbElement for publish requests.
|
|
"""
|
|
if self.items:
|
|
for item in self.items:
|
|
item.uri = NS_PUBSUB
|
|
verbElement.addChild(item)
|
|
|
|
|
|
def _parse_jid(self, verbElement):
|
|
"""
|
|
Parse subscriber out of the verbElement for un-/subscribe requests.
|
|
"""
|
|
try:
|
|
self.subscriber = jid.internJID(verbElement["jid"])
|
|
except KeyError:
|
|
raise BadRequest('jid-required')
|
|
|
|
|
|
def _render_jid(self, verbElement):
|
|
"""
|
|
Render subscriber into the verbElement for un-/subscribe requests.
|
|
"""
|
|
verbElement['jid'] = self.subscriber.full()
|
|
|
|
|
|
def _parse_default(self, verbElement):
|
|
"""
|
|
Parse node type out of a request for the default node configuration.
|
|
"""
|
|
form = data_form.findForm(verbElement, NS_PUBSUB_NODE_CONFIG)
|
|
if form is not None and form.formType == 'submit':
|
|
values = form.getValues()
|
|
self.nodeType = values.get('pubsub#node_type', 'leaf')
|
|
else:
|
|
self.nodeType = 'leaf'
|
|
|
|
|
|
def _parse_configure(self, verbElement):
|
|
"""
|
|
Parse options out of a request for setting the node configuration.
|
|
"""
|
|
form = data_form.findForm(verbElement, NS_PUBSUB_NODE_CONFIG)
|
|
if form is not None:
|
|
if form.formType in ('submit', 'cancel'):
|
|
self.options = form
|
|
else:
|
|
raise BadRequest(text=u"Unexpected form type '%s'" % form.formType)
|
|
else:
|
|
raise BadRequest(text="Missing configuration form")
|
|
|
|
|
|
def _parse_configureOrNone(self, verbElement):
|
|
"""
|
|
Parse optional node configuration form in create request.
|
|
"""
|
|
for element in verbElement.parent.elements():
|
|
if element.uri == NS_PUBSUB_OWNER and element.name == 'configure':
|
|
form = data_form.findForm(element, NS_PUBSUB_NODE_CONFIG)
|
|
if form is not None:
|
|
if form.formType != 'submit':
|
|
raise BadRequest(text=u"Unexpected form type '%s'" %
|
|
form.formType)
|
|
else:
|
|
form = data_form.Form('submit',
|
|
formNamespace=NS_PUBSUB_NODE_CONFIG)
|
|
self.options = form
|
|
|
|
|
|
def _render_configureOrNone(self, verbElement):
|
|
"""
|
|
Render optional node configuration form in create request.
|
|
"""
|
|
if self.options is not None:
|
|
if verbElement.name == 'configure':
|
|
configure = verbElement
|
|
else:
|
|
configure = verbElement.parent.addElement('configure')
|
|
configure.addChild(self.options.toElement())
|
|
|
|
|
|
def _parse_itemIdentifiers(self, verbElement):
|
|
"""
|
|
Parse item identifiers out of items and retract requests.
|
|
"""
|
|
self.itemIdentifiers = []
|
|
for element in verbElement.elements():
|
|
if element.uri == NS_PUBSUB and element.name == 'item':
|
|
try:
|
|
self.itemIdentifiers.append(element["id"])
|
|
except KeyError:
|
|
raise BadRequest()
|
|
|
|
|
|
def _render_itemIdentifiers(self, verbElement):
|
|
"""
|
|
Render item identifiers into items and retract requests.
|
|
"""
|
|
if self.itemIdentifiers:
|
|
for itemIdentifier in self.itemIdentifiers:
|
|
item = verbElement.addElement('item')
|
|
item['id'] = itemIdentifier
|
|
|
|
|
|
def _parse_maxItems(self, verbElement):
|
|
"""
|
|
Parse maximum items out of an items request.
|
|
"""
|
|
value = verbElement.getAttribute('max_items')
|
|
|
|
if value:
|
|
try:
|
|
self.maxItems = int(value)
|
|
except ValueError:
|
|
raise BadRequest(text="Field max_items requires a positive " +
|
|
"integer value")
|
|
|
|
|
|
def _render_maxItems(self, verbElement):
|
|
"""
|
|
Render maximum items into an items request.
|
|
"""
|
|
if self.maxItems:
|
|
verbElement['max_items'] = unicode(self.maxItems)
|
|
|
|
|
|
def _parse_subidOrNone(self, verbElement):
|
|
"""
|
|
Parse subscription identifier out of a request.
|
|
"""
|
|
self.subscriptionIdentifier = verbElement.getAttribute("subid")
|
|
|
|
|
|
def _render_subidOrNone(self, verbElement):
|
|
"""
|
|
Render subscription identifier into a request.
|
|
"""
|
|
if self.subscriptionIdentifier:
|
|
verbElement['subid'] = self.subscriptionIdentifier
|
|
|
|
|
|
def _parse_options(self, verbElement):
|
|
"""
|
|
Parse options form out of a subscription options request.
|
|
"""
|
|
form = data_form.findForm(verbElement, NS_PUBSUB_SUBSCRIBE_OPTIONS)
|
|
if form is not None:
|
|
if form.formType in ('submit', 'cancel'):
|
|
self.options = form
|
|
else:
|
|
raise BadRequest(text=u"Unexpected form type '%s'" % form.formType)
|
|
else:
|
|
raise BadRequest(text="Missing options form")
|
|
|
|
|
|
|
|
def _render_options(self, verbElement):
|
|
verbElement.addChild(self.options.toElement())
|
|
|
|
|
|
def _parse_optionsWithSubscribe(self, verbElement):
|
|
for element in verbElement.parent.elements():
|
|
if element.name == 'options' and element.uri == NS_PUBSUB:
|
|
form = data_form.findForm(element,
|
|
NS_PUBSUB_SUBSCRIBE_OPTIONS)
|
|
if form is not None:
|
|
if form.formType != 'submit':
|
|
raise BadRequest(text=u"Unexpected form type '%s'" %
|
|
form.formType)
|
|
else:
|
|
form = data_form.Form('submit',
|
|
formNamespace=NS_PUBSUB_SUBSCRIBE_OPTIONS)
|
|
self.options = form
|
|
|
|
|
|
def _render_optionsWithSubscribe(self, verbElement):
|
|
if self.options is not None:
|
|
optionsElement = verbElement.parent.addElement('options')
|
|
self._render_options(optionsElement)
|
|
|
|
|
|
def _parse_affiliations(self, verbElement):
|
|
self.affiliations = {}
|
|
for element in verbElement.elements():
|
|
if (element.uri == NS_PUBSUB_OWNER and
|
|
element.name == 'affiliation'):
|
|
try:
|
|
entity = jid.internJID(element['jid']).userhostJID()
|
|
except KeyError:
|
|
raise BadRequest(text='Missing jid attribute')
|
|
|
|
if entity in self.affiliations:
|
|
raise BadRequest(text='Multiple affiliations for an entity')
|
|
|
|
try:
|
|
affiliation = element['affiliation']
|
|
except KeyError:
|
|
raise BadRequest(text='Missing affiliation attribute')
|
|
|
|
self.affiliations[entity] = affiliation
|
|
|
|
|
|
def _parse_notify(self, verbElement):
|
|
value = verbElement.getAttribute('notify')
|
|
|
|
if value:
|
|
if value in BOOL_TRUE:
|
|
self.notify = True
|
|
elif value in BOOL_FALSE:
|
|
self.notify = False
|
|
else:
|
|
raise BadRequest(text="Field notify must be a boolean value")
|
|
|
|
|
|
def _render_notify(self, verbElement):
|
|
if self.notify is not None:
|
|
verbElement['notify'] = "true" if self.notify else "false"
|
|
|
|
|
|
def parseElement(self, element):
|
|
"""
|
|
Parse the publish-subscribe verb and parameters out of a request.
|
|
"""
|
|
generic.Stanza.parseElement(self, element)
|
|
|
|
verbs = []
|
|
verbElements = []
|
|
for child in element.pubsub.elements():
|
|
key = (self.stanzaType, child.uri, child.name)
|
|
try:
|
|
verb = self._requestVerbMap[key]
|
|
except KeyError:
|
|
continue
|
|
|
|
verbs.append(verb)
|
|
verbElements.append(child)
|
|
|
|
if not verbs:
|
|
raise NotImplementedError()
|
|
|
|
if len(verbs) > 1:
|
|
if 'optionsSet' in verbs and 'subscribe' in verbs:
|
|
self.verb = 'subscribe'
|
|
verbElement = verbElements[verbs.index('subscribe')]
|
|
else:
|
|
raise NotImplementedError()
|
|
else:
|
|
self.verb = verbs[0]
|
|
verbElement = verbElements[0]
|
|
|
|
for parameter in self._parameters[self.verb]:
|
|
getattr(self, '_parse_%s' % parameter)(verbElement)
|
|
|
|
|
|
|
|
def send(self, xs):
|
|
"""
|
|
Send this request to its recipient.
|
|
|
|
This renders all of the relevant parameters for this specific
|
|
requests into an L{IQ}, and invoke its C{send} method.
|
|
This returns a deferred that fires upon reception of a response. See
|
|
L{IQ} for details.
|
|
|
|
@param xs: The XML stream to send the request on.
|
|
@type xs: L{twisted.words.protocols.jabber.xmlstream.XmlStream}
|
|
@rtype: L{defer.Deferred}.
|
|
"""
|
|
|
|
try:
|
|
(self.stanzaType,
|
|
childURI,
|
|
childName) = self._verbRequestMap[self.verb]
|
|
except KeyError:
|
|
raise NotImplementedError()
|
|
|
|
iq = IQ(xs, self.stanzaType)
|
|
iq.addElement((childURI, 'pubsub'))
|
|
verbElement = iq.pubsub.addElement(childName)
|
|
|
|
if self.sender:
|
|
iq['from'] = self.sender.full()
|
|
if self.recipient:
|
|
iq['to'] = self.recipient.full()
|
|
|
|
for parameter in self._parameters[self.verb]:
|
|
getattr(self, '_render_%s' % parameter)(verbElement)
|
|
|
|
return iq.send()
|
|
|
|
|
|
|
|
class PubSubEvent(object):
|
|
"""
|
|
A publish subscribe event.
|
|
|
|
@param sender: The entity from which the notification was received.
|
|
@type sender: L{jid.JID}
|
|
@param recipient: The entity to which the notification was sent.
|
|
@type recipient: L{wokkel.pubsub.ItemsEvent}
|
|
@param nodeIdentifier: Identifier of the node the event pertains to.
|
|
@type nodeIdentifier: C{unicode}
|
|
@param headers: SHIM headers, see L{wokkel.shim.extractHeaders}.
|
|
@type headers: C{dict}
|
|
"""
|
|
|
|
def __init__(self, sender, recipient, nodeIdentifier, headers):
|
|
self.sender = sender
|
|
self.recipient = recipient
|
|
self.nodeIdentifier = nodeIdentifier
|
|
self.headers = headers
|
|
|
|
|
|
|
|
class ItemsEvent(PubSubEvent):
|
|
"""
|
|
A publish-subscribe event that signifies new, updated and retracted items.
|
|
|
|
@param items: List of received items as domish elements.
|
|
@type items: C{list} of L{domish.Element}
|
|
"""
|
|
|
|
def __init__(self, sender, recipient, nodeIdentifier, items, headers):
|
|
PubSubEvent.__init__(self, sender, recipient, nodeIdentifier, headers)
|
|
self.items = items
|
|
|
|
|
|
|
|
class DeleteEvent(PubSubEvent):
|
|
"""
|
|
A publish-subscribe event that signifies the deletion of a node.
|
|
"""
|
|
|
|
redirectURI = None
|
|
|
|
|
|
|
|
class PurgeEvent(PubSubEvent):
|
|
"""
|
|
A publish-subscribe event that signifies the purging of a node.
|
|
"""
|
|
|
|
|
|
|
|
class PubSubClient(XMPPHandler):
|
|
"""
|
|
Publish subscribe client protocol.
|
|
"""
|
|
implements(IPubSubClient)
|
|
|
|
_request_class = PubSubRequest
|
|
|
|
def connectionInitialized(self):
|
|
self.xmlstream.addObserver('/message/event[@xmlns="%s"]' %
|
|
NS_PUBSUB_EVENT, self._onEvent)
|
|
|
|
|
|
def _onEvent(self, message):
|
|
if message.getAttribute('type') == 'error':
|
|
return
|
|
|
|
try:
|
|
sender = jid.JID(message["from"])
|
|
recipient = jid.JID(message["to"])
|
|
except KeyError:
|
|
return
|
|
|
|
actionElement = None
|
|
for element in message.event.elements():
|
|
if element.uri == NS_PUBSUB_EVENT:
|
|
actionElement = element
|
|
|
|
if not actionElement:
|
|
return
|
|
|
|
eventHandler = getattr(self, "_onEvent_%s" % actionElement.name, None)
|
|
|
|
if eventHandler:
|
|
headers = shim.extractHeaders(message)
|
|
eventHandler(sender, recipient, actionElement, headers)
|
|
message.handled = True
|
|
|
|
|
|
def _onEvent_items(self, sender, recipient, action, headers):
|
|
nodeIdentifier = action["node"]
|
|
|
|
items = [element for element in action.elements()
|
|
if element.name in ('item', 'retract')]
|
|
|
|
event = ItemsEvent(sender, recipient, nodeIdentifier, items, headers)
|
|
self.itemsReceived(event)
|
|
|
|
|
|
def _onEvent_delete(self, sender, recipient, action, headers):
|
|
nodeIdentifier = action["node"]
|
|
event = DeleteEvent(sender, recipient, nodeIdentifier, headers)
|
|
if action.redirect:
|
|
event.redirectURI = action.redirect.getAttribute('uri')
|
|
self.deleteReceived(event)
|
|
|
|
|
|
def _onEvent_purge(self, sender, recipient, action, headers):
|
|
nodeIdentifier = action["node"]
|
|
event = PurgeEvent(sender, recipient, nodeIdentifier, headers)
|
|
self.purgeReceived(event)
|
|
|
|
|
|
def itemsReceived(self, event):
|
|
pass
|
|
|
|
|
|
def deleteReceived(self, event):
|
|
pass
|
|
|
|
|
|
def purgeReceived(self, event):
|
|
pass
|
|
|
|
|
|
def createNode(self, service, nodeIdentifier=None, options=None,
|
|
sender=None):
|
|
"""
|
|
Create a publish subscribe node.
|
|
|
|
@param service: The publish subscribe service to create the node at.
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
@param nodeIdentifier: Optional suggestion for the id of the node.
|
|
@type nodeIdentifier: C{unicode}
|
|
@param options: Optional node configuration options.
|
|
@type options: C{dict}
|
|
"""
|
|
request = self._request_class('create')
|
|
request.recipient = service
|
|
request.nodeIdentifier = nodeIdentifier
|
|
request.sender = sender
|
|
|
|
if options:
|
|
form = data_form.Form(formType='submit',
|
|
formNamespace=NS_PUBSUB_NODE_CONFIG)
|
|
form.makeFields(options)
|
|
request.options = form
|
|
|
|
def cb(iq):
|
|
try:
|
|
new_node = iq.pubsub.create["node"]
|
|
except AttributeError:
|
|
# the suggested node identifier was accepted
|
|
new_node = nodeIdentifier
|
|
return new_node
|
|
|
|
d = request.send(self.xmlstream)
|
|
d.addCallback(cb)
|
|
return d
|
|
|
|
|
|
def deleteNode(self, service, nodeIdentifier, sender=None):
|
|
"""
|
|
Delete a publish subscribe node.
|
|
|
|
@param service: The publish subscribe service to delete the node from.
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
@param nodeIdentifier: The identifier of the node.
|
|
@type nodeIdentifier: C{unicode}
|
|
"""
|
|
request = self._request_class('delete')
|
|
request.recipient = service
|
|
request.nodeIdentifier = nodeIdentifier
|
|
request.sender = sender
|
|
return request.send(self.xmlstream)
|
|
|
|
|
|
def subscribe(self, service, nodeIdentifier, subscriber,
|
|
options=None, sender=None):
|
|
"""
|
|
Subscribe to a publish subscribe node.
|
|
|
|
@param service: The publish subscribe service that keeps the node.
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param nodeIdentifier: The identifier of the node.
|
|
@type nodeIdentifier: C{unicode}
|
|
|
|
@param subscriber: The entity to subscribe to the node. This entity
|
|
will get notifications of new published items.
|
|
@type subscriber: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param options: Subscription options.
|
|
@type options: C{dict}
|
|
|
|
@return: Deferred that fires with L{Subscription} or errbacks with
|
|
L{SubscriptionPending} or L{SubscriptionUnconfigured}.
|
|
@rtype: L{defer.Deferred}
|
|
"""
|
|
request = self._request_class('subscribe')
|
|
request.recipient = service
|
|
request.nodeIdentifier = nodeIdentifier
|
|
request.subscriber = subscriber
|
|
request.sender = sender
|
|
|
|
if options:
|
|
form = data_form.Form(formType='submit',
|
|
formNamespace=NS_PUBSUB_SUBSCRIBE_OPTIONS)
|
|
form.makeFields(options)
|
|
request.options = form
|
|
|
|
def cb(iq):
|
|
subscription = Subscription.fromElement(iq.pubsub.subscription)
|
|
|
|
if subscription.state == 'pending':
|
|
raise SubscriptionPending()
|
|
elif subscription.state == 'unconfigured':
|
|
raise SubscriptionUnconfigured()
|
|
else:
|
|
# we assume subscription == 'subscribed'
|
|
# any other value would be invalid, but that should have
|
|
# yielded a stanza error.
|
|
return subscription
|
|
|
|
d = request.send(self.xmlstream)
|
|
d.addCallback(cb)
|
|
return d
|
|
|
|
|
|
def unsubscribe(self, service, nodeIdentifier, subscriber,
|
|
subscriptionIdentifier=None, sender=None):
|
|
"""
|
|
Unsubscribe from a publish subscribe node.
|
|
|
|
@param service: The publish subscribe service that keeps the node.
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param nodeIdentifier: The identifier of the node.
|
|
@type nodeIdentifier: C{unicode}
|
|
|
|
@param subscriber: The entity to unsubscribe from the node.
|
|
@type subscriber: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param subscriptionIdentifier: Optional subscription identifier.
|
|
@type subscriptionIdentifier: C{unicode}
|
|
"""
|
|
request = self._request_class('unsubscribe')
|
|
request.recipient = service
|
|
request.nodeIdentifier = nodeIdentifier
|
|
request.subscriber = subscriber
|
|
request.subscriptionIdentifier = subscriptionIdentifier
|
|
request.sender = sender
|
|
return request.send(self.xmlstream)
|
|
|
|
|
|
def publish(self, service, nodeIdentifier, items=None, sender=None):
|
|
"""
|
|
Publish to a publish subscribe node.
|
|
|
|
@param service: The publish subscribe service that keeps the node.
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
@param nodeIdentifier: The identifier of the node.
|
|
@type nodeIdentifier: C{unicode}
|
|
@param items: Optional list of L{Item}s to publish.
|
|
@type items: C{list}
|
|
"""
|
|
request = self._request_class('publish')
|
|
request.recipient = service
|
|
request.nodeIdentifier = nodeIdentifier
|
|
request.items = items
|
|
request.sender = sender
|
|
return request.send(self.xmlstream)
|
|
|
|
|
|
def items(self, service, nodeIdentifier, maxItems=None, itemIdentifiers=None,
|
|
subscriptionIdentifier=None, sender=None):
|
|
"""
|
|
Retrieve previously published items from a publish subscribe node.
|
|
|
|
@param service: The publish subscribe service that keeps the node.
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param nodeIdentifier: The identifier of the node.
|
|
@type nodeIdentifier: C{unicode}
|
|
|
|
@param maxItems: Optional limit on the number of retrieved items.
|
|
@type maxItems: C{int}
|
|
|
|
@param itemIdentifiers: Identifiers of the items to be retrieved.
|
|
@type itemIdentifiers: C{set}
|
|
|
|
@param subscriptionIdentifier: Optional subscription identifier. In
|
|
case the node has been subscribed to multiple times, this narrows
|
|
the results to the specific subscription.
|
|
@type subscriptionIdentifier: C{unicode}
|
|
"""
|
|
request = self._request_class('items')
|
|
request.recipient = service
|
|
request.nodeIdentifier = nodeIdentifier
|
|
if maxItems:
|
|
request.maxItems = str(int(maxItems))
|
|
request.subscriptionIdentifier = subscriptionIdentifier
|
|
request.sender = sender
|
|
request.itemIdentifiers = itemIdentifiers
|
|
|
|
def cb(iq):
|
|
items = []
|
|
for element in iq.pubsub.items.elements():
|
|
if element.uri == NS_PUBSUB and element.name == 'item':
|
|
items.append(element)
|
|
return items
|
|
|
|
d = request.send(self.xmlstream)
|
|
d.addCallback(cb)
|
|
return d
|
|
|
|
def retractItems(self, service, nodeIdentifier, itemIdentifiers, notify=None, sender=None):
|
|
"""
|
|
Retract items from a publish subscribe node.
|
|
|
|
@param service: The publish subscribe service to delete the node from.
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
@param nodeIdentifier: The identifier of the node.
|
|
@type nodeIdentifier: C{unicode}
|
|
@param itemIdentifiers: Identifiers of the items to be retracted.
|
|
@type itemIdentifiers: C{set}
|
|
@param notify: True if notification is required
|
|
@type notify: C{unicode}
|
|
"""
|
|
request = self._request_class('retract')
|
|
request.recipient = service
|
|
request.nodeIdentifier = nodeIdentifier
|
|
request.itemIdentifiers = itemIdentifiers
|
|
request.notify = notify
|
|
request.sender = sender
|
|
return request.send(self.xmlstream)
|
|
|
|
def getOptions(self, service, nodeIdentifier, subscriber,
|
|
subscriptionIdentifier=None, sender=None):
|
|
"""
|
|
Get subscription options.
|
|
|
|
@param service: The publish subscribe service that keeps the node.
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param nodeIdentifier: The identifier of the node.
|
|
@type nodeIdentifier: C{unicode}
|
|
|
|
@param subscriber: The entity subscribed to the node.
|
|
@type subscriber: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param subscriptionIdentifier: Optional subscription identifier.
|
|
@type subscriptionIdentifier: C{unicode}
|
|
|
|
@rtype: L{data_form.Form}
|
|
"""
|
|
request = self._request_class('optionsGet')
|
|
request.recipient = service
|
|
request.nodeIdentifier = nodeIdentifier
|
|
request.subscriber = subscriber
|
|
request.subscriptionIdentifier = subscriptionIdentifier
|
|
request.sender = sender
|
|
|
|
def cb(iq):
|
|
form = data_form.findForm(iq.pubsub.options,
|
|
NS_PUBSUB_SUBSCRIBE_OPTIONS)
|
|
form.typeCheck()
|
|
return form
|
|
|
|
d = request.send(self.xmlstream)
|
|
d.addCallback(cb)
|
|
return d
|
|
|
|
|
|
def setOptions(self, service, nodeIdentifier, subscriber,
|
|
options, subscriptionIdentifier=None, sender=None):
|
|
"""
|
|
Set subscription options.
|
|
|
|
@param service: The publish subscribe service that keeps the node.
|
|
@type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param nodeIdentifier: The identifier of the node.
|
|
@type nodeIdentifier: C{unicode}
|
|
|
|
@param subscriber: The entity subscribed to the node.
|
|
@type subscriber: L{JID<twisted.words.protocols.jabber.jid.JID>}
|
|
|
|
@param options: Subscription options.
|
|
@type options: C{dict}.
|
|
|
|
@param subscriptionIdentifier: Optional subscription identifier.
|
|
@type subscriptionIdentifier: C{unicode}
|
|
"""
|
|
request = self._request_class('optionsSet')
|
|
request.recipient = service
|
|
request.nodeIdentifier = nodeIdentifier
|
|
request.subscriber = subscriber
|
|
request.subscriptionIdentifier = subscriptionIdentifier
|
|
request.sender = sender
|
|
|
|
form = data_form.Form(formType='submit',
|
|
formNamespace=NS_PUBSUB_SUBSCRIBE_OPTIONS)
|
|
form.makeFields(options)
|
|
request.options = form
|
|
|
|
d = request.send(self.xmlstream)
|
|
return d
|
|
|
|
|
|
|
|
class PubSubService(XMPPHandler, IQHandlerMixin):
|
|
"""
|
|
Protocol implementation for a XMPP Publish Subscribe Service.
|
|
|
|
The word Service here is used as taken from the Publish Subscribe
|
|
specification. It is the party responsible for keeping nodes and their
|
|
subscriptions, and sending out notifications.
|
|
|
|
Methods from the L{IPubSubService} interface that are called as a result
|
|
of an XMPP request may raise exceptions. Alternatively the deferred
|
|
returned by these methods may have their errback called. These are handled
|
|
as follows:
|
|
|
|
- If the exception is an instance of L{error.StanzaError}, an error
|
|
response iq is returned.
|
|
- Any other exception is reported using L{log.msg}. An error response
|
|
with the condition C{internal-server-error} is returned.
|
|
|
|
The default implementation of said methods raises an L{Unsupported}
|
|
exception and are meant to be overridden.
|
|
|
|
@ivar discoIdentity: Service discovery identity as a dictionary with
|
|
keys C{'category'}, C{'type'} and C{'name'}.
|
|
@ivar pubSubFeatures: List of supported publish-subscribe features for
|
|
service discovery, as C{str}.
|
|
@type pubSubFeatures: C{list} or C{None}
|
|
"""
|
|
|
|
implements(IPubSubService, disco.IDisco)
|
|
|
|
iqHandlers = {
|
|
'/*': '_onPubSubRequest',
|
|
}
|
|
|
|
_legacyHandlers = {
|
|
'publish': ('publish', ['sender', 'recipient',
|
|
'nodeIdentifier', 'items']),
|
|
'subscribe': ('subscribe', ['sender', 'recipient',
|
|
'nodeIdentifier', 'subscriber']),
|
|
'unsubscribe': ('unsubscribe', ['sender', 'recipient',
|
|
'nodeIdentifier', 'subscriber']),
|
|
'subscriptions': ('subscriptions', ['sender', 'recipient']),
|
|
'affiliations': ('affiliations', ['sender', 'recipient']),
|
|
'create': ('create', ['sender', 'recipient', 'nodeIdentifier']),
|
|
'getConfigurationOptions': ('getConfigurationOptions', []),
|
|
'default': ('getDefaultConfiguration',
|
|
['sender', 'recipient', 'nodeType']),
|
|
'configureGet': ('getConfiguration', ['sender', 'recipient',
|
|
'nodeIdentifier']),
|
|
'configureSet': ('setConfiguration', ['sender', 'recipient',
|
|
'nodeIdentifier', 'options']),
|
|
'items': ('items', ['sender', 'recipient', 'nodeIdentifier',
|
|
'maxItems', 'itemIdentifiers']),
|
|
'retract': ('retract', ['sender', 'recipient', 'nodeIdentifier',
|
|
'itemIdentifiers']),
|
|
'purge': ('purge', ['sender', 'recipient', 'nodeIdentifier']),
|
|
'delete': ('delete', ['sender', 'recipient', 'nodeIdentifier']),
|
|
}
|
|
|
|
_request_class = PubSubRequest
|
|
|
|
hideNodes = False
|
|
|
|
def __init__(self, resource=None):
|
|
self.resource = resource
|
|
self.discoIdentity = {'category': 'pubsub',
|
|
'type': 'service',
|
|
'name': 'Generic Publish-Subscribe Service'}
|
|
|
|
self.pubSubFeatures = []
|
|
|
|
|
|
def connectionMade(self):
|
|
self.xmlstream.addObserver(PUBSUB_REQUEST, self.handleRequest)
|
|
|
|
|
|
def getDiscoInfo(self, requestor, target, nodeIdentifier=''):
|
|
def toInfo(nodeInfo):
|
|
if not nodeInfo:
|
|
return
|
|
|
|
(nodeType, metaData) = nodeInfo['type'], nodeInfo['meta-data']
|
|
info.append(disco.DiscoIdentity('pubsub', nodeType))
|
|
if metaData:
|
|
form = data_form.Form(formType="result",
|
|
formNamespace=NS_PUBSUB_META_DATA)
|
|
form.addField(
|
|
data_form.Field(
|
|
var='pubsub#node_type',
|
|
value=nodeType,
|
|
label='The type of node (collection or leaf)'
|
|
)
|
|
)
|
|
|
|
for metaDatum in metaData:
|
|
form.addField(data_form.Field.fromDict(metaDatum))
|
|
|
|
info.append(form)
|
|
|
|
return
|
|
|
|
info = []
|
|
|
|
request = self._request_class('discoInfo')
|
|
|
|
if self.resource is not None:
|
|
resource = self.resource.locateResource(request)
|
|
identity = resource.discoIdentity
|
|
features = resource.features
|
|
getInfo = resource.getInfo
|
|
else:
|
|
category = self.discoIdentity['category']
|
|
idType = self.discoIdentity['type']
|
|
name = self.discoIdentity['name']
|
|
identity = disco.DiscoIdentity(category, idType, name)
|
|
features = self.pubSubFeatures
|
|
getInfo = self.getNodeInfo
|
|
|
|
if not nodeIdentifier:
|
|
info.append(identity)
|
|
info.append(disco.DiscoFeature(disco.NS_DISCO_ITEMS))
|
|
info.extend([disco.DiscoFeature("%s#%s" % (NS_PUBSUB, feature))
|
|
for feature in features])
|
|
|
|
d = defer.maybeDeferred(getInfo, requestor, target, nodeIdentifier or '')
|
|
d.addCallback(toInfo)
|
|
d.addErrback(log.err)
|
|
d.addCallback(lambda _: info)
|
|
return d
|
|
|
|
|
|
def getDiscoItems(self, requestor, target, nodeIdentifier=''):
|
|
if self.hideNodes:
|
|
d = defer.succeed([])
|
|
elif self.resource is not None:
|
|
request = self._request_class('discoInfo')
|
|
resource = self.resource.locateResource(request)
|
|
d = resource.getNodes(requestor, target, nodeIdentifier)
|
|
elif nodeIdentifier:
|
|
d = self.getNodes(requestor, target)
|
|
else:
|
|
d = defer.succeed([])
|
|
|
|
d.addCallback(lambda nodes: [disco.DiscoItem(target, node)
|
|
for node in nodes])
|
|
return d
|
|
|
|
|
|
def _onPubSubRequest(self, iq):
|
|
request = self._request_class.fromElement(iq)
|
|
if self.resource is not None:
|
|
resource = self.resource.locateResource(request)
|
|
else:
|
|
resource = self
|
|
|
|
# Preprocess the request, knowing the handling resource
|
|
try:
|
|
preProcessor = getattr(self, '_preProcess_%s' % request.verb)
|
|
except AttributeError:
|
|
pass
|
|
else:
|
|
request = preProcessor(resource, request)
|
|
if request is None:
|
|
return defer.succeed(None)
|
|
|
|
# Process the request itself,
|
|
if resource is not self:
|
|
try:
|
|
handler = getattr(resource, request.verb)
|
|
except AttributeError:
|
|
text = "Request verb: %s" % request.verb
|
|
return defer.fail(Unsupported('', text))
|
|
|
|
d = handler(request)
|
|
else:
|
|
try:
|
|
handlerName, argNames = self._legacyHandlers[request.verb]
|
|
except KeyError:
|
|
text = "Request verb: %s" % request.verb
|
|
return defer.fail(Unsupported('', text))
|
|
|
|
handler = getattr(self, handlerName)
|
|
args = [getattr(request, arg) for arg in argNames]
|
|
d = handler(*args)
|
|
|
|
# If needed, translate the result into a response
|
|
try:
|
|
cb = getattr(self, '_toResponse_%s' % request.verb)
|
|
except AttributeError:
|
|
pass
|
|
else:
|
|
d.addCallback(cb, resource, request)
|
|
|
|
return d
|
|
|
|
|
|
def _toResponse_subscribe(self, result, resource, request):
|
|
response = domish.Element((NS_PUBSUB, "pubsub"))
|
|
response.addChild(result.toElement(NS_PUBSUB))
|
|
return response
|
|
|
|
|
|
def _toResponse_subscriptions(self, result, resource, request):
|
|
response = domish.Element((NS_PUBSUB, 'pubsub'))
|
|
subscriptions = response.addElement('subscriptions')
|
|
for subscription in result:
|
|
subscriptions.addChild(subscription.toElement(NS_PUBSUB))
|
|
return response
|
|
|
|
|
|
def _toResponse_affiliations(self, result, resource, request):
|
|
response = domish.Element((NS_PUBSUB, 'pubsub'))
|
|
affiliations = response.addElement('affiliations')
|
|
|
|
for nodeIdentifier, affiliation in result:
|
|
item = affiliations.addElement('affiliation')
|
|
item['node'] = nodeIdentifier
|
|
item['affiliation'] = affiliation
|
|
|
|
return response
|
|
|
|
|
|
def _toResponse_create(self, result, resource, request):
|
|
if not request.nodeIdentifier or request.nodeIdentifier != result:
|
|
response = domish.Element((NS_PUBSUB, 'pubsub'))
|
|
create = response.addElement('create')
|
|
create['node'] = result
|
|
return response
|
|
else:
|
|
return None
|
|
|
|
|
|
def _formFromConfiguration(self, resource, values):
|
|
fieldDefs = resource.getConfigurationOptions()
|
|
form = data_form.Form(formType="form",
|
|
formNamespace=NS_PUBSUB_NODE_CONFIG)
|
|
form.makeFields(values, fieldDefs)
|
|
return form
|
|
|
|
|
|
def _checkConfiguration(self, resource, form):
|
|
fieldDefs = resource.getConfigurationOptions()
|
|
form.typeCheck(fieldDefs, filterUnknown=True)
|
|
|
|
|
|
def _preProcess_create(self, resource, request):
|
|
if request.options:
|
|
self._checkConfiguration(resource, request.options)
|
|
return request
|
|
|
|
|
|
def _preProcess_default(self, resource, request):
|
|
if request.nodeType not in ('leaf', 'collection'):
|
|
raise error.StanzaError('not-acceptable')
|
|
else:
|
|
return request
|
|
|
|
|
|
def _toResponse_default(self, options, resource, request):
|
|
response = domish.Element((NS_PUBSUB_OWNER, "pubsub"))
|
|
default = response.addElement("default")
|
|
form = self._formFromConfiguration(resource, options)
|
|
default.addChild(form.toElement())
|
|
return response
|
|
|
|
|
|
def _toResponse_configureGet(self, options, resource, request):
|
|
response = domish.Element((NS_PUBSUB_OWNER, "pubsub"))
|
|
configure = response.addElement("configure")
|
|
form = self._formFromConfiguration(resource, options)
|
|
configure.addChild(form.toElement())
|
|
|
|
if request.nodeIdentifier:
|
|
configure["node"] = request.nodeIdentifier
|
|
|
|
return response
|
|
|
|
|
|
def _preProcess_configureSet(self, resource, request):
|
|
if request.options.formType == 'cancel':
|
|
return None
|
|
else:
|
|
self._checkConfiguration(resource, request.options)
|
|
return request
|
|
|
|
|
|
def _toResponse_items(self, result, resource, request):
|
|
response = domish.Element((NS_PUBSUB, 'pubsub'))
|
|
items = response.addElement('items')
|
|
items["node"] = request.nodeIdentifier
|
|
|
|
for item in result:
|
|
item.uri = NS_PUBSUB
|
|
items.addChild(item)
|
|
|
|
return response
|
|
|
|
|
|
def _createNotification(self, eventType, service, nodeIdentifier,
|
|
subscriber, subscriptions=None):
|
|
headers = []
|
|
|
|
if subscriptions:
|
|
for subscription in subscriptions:
|
|
if nodeIdentifier != subscription.nodeIdentifier:
|
|
headers.append(('Collection', subscription.nodeIdentifier))
|
|
|
|
message = domish.Element((None, "message"))
|
|
message["from"] = service.full()
|
|
message["to"] = subscriber.full()
|
|
event = message.addElement((NS_PUBSUB_EVENT, "event"))
|
|
|
|
element = event.addElement(eventType)
|
|
element["node"] = nodeIdentifier
|
|
|
|
if headers:
|
|
message.addChild(shim.Headers(headers))
|
|
|
|
return message
|
|
|
|
|
|
def _toResponse_affiliationsGet(self, result, resource, request):
|
|
response = domish.Element((NS_PUBSUB_OWNER, 'pubsub'))
|
|
affiliations = response.addElement('affiliations')
|
|
|
|
if request.nodeIdentifier:
|
|
affiliations['node'] = request.nodeIdentifier
|
|
|
|
for entity, affiliation in result.iteritems():
|
|
item = affiliations.addElement('affiliation')
|
|
item['jid'] = entity.full()
|
|
item['affiliation'] = affiliation
|
|
|
|
return response
|
|
|
|
|
|
# public methods
|
|
|
|
def notifyPublish(self, service, nodeIdentifier, notifications):
|
|
for subscriber, subscriptions, items in notifications:
|
|
message = self._createNotification('items', service,
|
|
nodeIdentifier, subscriber,
|
|
subscriptions)
|
|
for item in items:
|
|
item.uri = NS_PUBSUB_EVENT
|
|
message.event.items.addChild(item)
|
|
self.send(message)
|
|
|
|
|
|
def notifyRetract(self, service, nodeIdentifier, notifications):
|
|
for subscriber, subscriptions, items in notifications:
|
|
message = self._createNotification('items', service,
|
|
nodeIdentifier, subscriber,
|
|
subscriptions)
|
|
for item in items:
|
|
retract = domish.Element((None, "retract"))
|
|
retract['id'] = item['id']
|
|
message.event.items.addChild(retract)
|
|
self.send(message)
|
|
|
|
|
|
def notifyDelete(self, service, nodeIdentifier, subscribers,
|
|
redirectURI=None):
|
|
for subscriber in subscribers:
|
|
message = self._createNotification('delete', service,
|
|
nodeIdentifier,
|
|
subscriber)
|
|
if redirectURI:
|
|
redirect = message.event.delete.addElement('redirect')
|
|
redirect['uri'] = redirectURI
|
|
self.send(message)
|
|
|
|
|
|
def getNodeInfo(self, requestor, service, nodeIdentifier):
|
|
return None
|
|
|
|
|
|
def getNodes(self, requestor, service):
|
|
return []
|
|
|
|
|
|
def publish(self, requestor, service, nodeIdentifier, items):
|
|
raise Unsupported('publish')
|
|
|
|
|
|
def subscribe(self, requestor, service, nodeIdentifier, subscriber):
|
|
raise Unsupported('subscribe')
|
|
|
|
|
|
def unsubscribe(self, requestor, service, nodeIdentifier, subscriber):
|
|
raise Unsupported('subscribe')
|
|
|
|
|
|
def subscriptions(self, requestor, service):
|
|
raise Unsupported('retrieve-subscriptions')
|
|
|
|
|
|
def affiliations(self, requestor, service):
|
|
raise Unsupported('retrieve-affiliations')
|
|
|
|
|
|
def create(self, requestor, service, nodeIdentifier):
|
|
raise Unsupported('create-nodes')
|
|
|
|
|
|
def getConfigurationOptions(self):
|
|
return {}
|
|
|
|
|
|
def getDefaultConfiguration(self, requestor, service, nodeType):
|
|
raise Unsupported('retrieve-default')
|
|
|
|
|
|
def getConfiguration(self, requestor, service, nodeIdentifier):
|
|
raise Unsupported('config-node')
|
|
|
|
|
|
def setConfiguration(self, requestor, service, nodeIdentifier, options):
|
|
raise Unsupported('config-node')
|
|
|
|
|
|
def items(self, requestor, service, nodeIdentifier, maxItems,
|
|
itemIdentifiers):
|
|
raise Unsupported('retrieve-items')
|
|
|
|
|
|
def retract(self, requestor, service, nodeIdentifier, itemIdentifiers):
|
|
raise Unsupported('retract-items')
|
|
|
|
|
|
def purge(self, requestor, service, nodeIdentifier):
|
|
raise Unsupported('purge-nodes')
|
|
|
|
|
|
def delete(self, requestor, service, nodeIdentifier):
|
|
raise Unsupported('delete-nodes')
|
|
|
|
|
|
|
|
class PubSubResource(object):
|
|
|
|
implements(IPubSubResource)
|
|
|
|
features = []
|
|
discoIdentity = disco.DiscoIdentity('pubsub',
|
|
'service',
|
|
'Publish-Subscribe Service')
|
|
|
|
|
|
def locateResource(self, request):
|
|
return self
|
|
|
|
|
|
def getInfo(self, requestor, service, nodeIdentifier):
|
|
return defer.succeed(None)
|
|
|
|
|
|
def getNodes(self, requestor, service, nodeIdentifier):
|
|
return defer.succeed([])
|
|
|
|
|
|
def getConfigurationOptions(self):
|
|
return {}
|
|
|
|
|
|
def publish(self, request):
|
|
return defer.fail(Unsupported('publish'))
|
|
|
|
|
|
def subscribe(self, request):
|
|
return defer.fail(Unsupported('subscribe'))
|
|
|
|
|
|
def unsubscribe(self, request):
|
|
return defer.fail(Unsupported('subscribe'))
|
|
|
|
|
|
def subscriptions(self, request):
|
|
return defer.fail(Unsupported('retrieve-subscriptions'))
|
|
|
|
|
|
def affiliations(self, request):
|
|
return defer.fail(Unsupported('retrieve-affiliations'))
|
|
|
|
|
|
def create(self, request):
|
|
return defer.fail(Unsupported('create-nodes'))
|
|
|
|
|
|
def default(self, request):
|
|
return defer.fail(Unsupported('retrieve-default'))
|
|
|
|
|
|
def configureGet(self, request):
|
|
return defer.fail(Unsupported('config-node'))
|
|
|
|
|
|
def configureSet(self, request):
|
|
return defer.fail(Unsupported('config-node'))
|
|
|
|
|
|
def items(self, request):
|
|
return defer.fail(Unsupported('retrieve-items'))
|
|
|
|
|
|
def retract(self, request):
|
|
return defer.fail(Unsupported('retract-items'))
|
|
|
|
|
|
def purge(self, request):
|
|
return defer.fail(Unsupported('purge-nodes'))
|
|
|
|
|
|
def delete(self, request):
|
|
return defer.fail(Unsupported('delete-nodes'))
|
|
|
|
|
|
def affiliationsGet(self, request):
|
|
return defer.fail(Unsupported('modify-affiliations'))
|
|
|
|
|
|
def affiliationsSet(self, request):
|
|
return defer.fail(Unsupported('modify-affiliations'))
|
|
|
|
|
|
def subscriptionsGet(self, request):
|
|
return defer.fail(Unsupported('manage-subscriptions'))
|
|
|
|
|
|
def subscriptionsSet(self, request):
|
|
return defer.fail(Unsupported('manage-subscriptions'))
|