# Slixmpp: The Slick XMPP Library
# Copyright (C) 2011 Nathanael C. Fritz
# This file is part of Slixmpp.
# See the file LICENSE for copying permission.
from slixmpp import Iq, Message
from slixmpp.xmlstream import register_stanza_plugin, ElementBase, ET, JID
from slixmpp.plugins import xep_0004
from slixmpp.plugins.xep_0060.stanza.base import OptionalSetting
[docs]
class Pubsub(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'pubsub'
plugin_attrib = name
interfaces = set(tuple())
[docs]
class Affiliations(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'affiliations'
plugin_attrib = name
interfaces = {'node'}
[docs]
class Affiliation(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'affiliation'
plugin_attrib = name
interfaces = {'node', 'affiliation', 'jid'}
[docs]
def set_jid(self, value):
self._set_attr('jid', str(value))
[docs]
def get_jid(self):
return JID(self._get_attr('jid'))
[docs]
class Subscription(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'subscription'
plugin_attrib = name
interfaces = {'jid', 'node', 'subscription', 'subid'}
[docs]
def set_jid(self, value):
self._set_attr('jid', str(value))
[docs]
def get_jid(self):
return JID(self._get_attr('jid'))
[docs]
class Subscriptions(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'subscriptions'
plugin_attrib = name
interfaces = {'node'}
[docs]
class SubscribeOptions(ElementBase, OptionalSetting):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'subscribe-options'
plugin_attrib = 'suboptions'
interfaces = {'required'}
[docs]
class Item(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'item'
plugin_attrib = name
interfaces = {'id', 'payload'}
[docs]
def set_payload(self, value):
del self['payload']
if isinstance(value, ElementBase):
if value.tag_name() in self.plugin_tag_map:
self.init_plugin(value.plugin_attrib, existing_xml=value.xml)
self.xml.append(value.xml)
else:
self.xml.append(value)
[docs]
def get_payload(self):
children = list(self.xml)
if len(children) > 0:
return children[0]
[docs]
def del_payload(self):
for child in self.xml:
self.xml.remove(child)
[docs]
class Items(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'items'
plugin_attrib = name
interfaces = {'node', 'max_items'}
[docs]
def set_max_items(self, value):
self._set_attr('max_items', str(value))
[docs]
class Create(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'create'
plugin_attrib = name
interfaces = {'node'}
[docs]
class Default(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'default'
plugin_attrib = name
interfaces = {'node', 'type'}
[docs]
def get_type(self):
t = self._get_attr('type')
if not t:
return 'leaf'
return t
[docs]
class Publish(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'publish'
plugin_attrib = name
interfaces = {'node'}
[docs]
class Retract(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'retract'
plugin_attrib = name
interfaces = {'node', 'notify'}
[docs]
def get_notify(self):
notify = self._get_attr('notify')
if notify in ('0', 'false'):
return False
elif notify in ('1', 'true'):
return True
return None
[docs]
def set_notify(self, value):
del self['notify']
if value is None:
return
elif value in (True, '1', 'true', 'True'):
self._set_attr('notify', 'true')
else:
self._set_attr('notify', 'false')
[docs]
class Unsubscribe(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'unsubscribe'
plugin_attrib = name
interfaces = {'node', 'jid', 'subid'}
[docs]
def set_jid(self, value):
self._set_attr('jid', str(value))
[docs]
def get_jid(self):
return JID(self._get_attr('jid'))
[docs]
class Subscribe(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'subscribe'
plugin_attrib = name
interfaces = {'node', 'jid'}
[docs]
def set_jid(self, value):
self._set_attr('jid', str(value))
[docs]
def get_jid(self):
return JID(self._get_attr('jid'))
[docs]
class Options(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'options'
plugin_attrib = name
interfaces = {'jid', 'node', 'options'}
def __init__(self, *args, **kwargs):
ElementBase.__init__(self, *args, **kwargs)
[docs]
def get_options(self):
config = self.xml.find('{jabber:x:data}x')
form = xep_0004.Form(xml=config)
return form
[docs]
def set_options(self, value):
if isinstance(value, ElementBase):
self.xml.append(value.xml)
else:
self.xml.append(value)
return self
[docs]
def del_options(self):
config = self.xml.find('{jabber:x:data}x')
self.xml.remove(config)
[docs]
def set_jid(self, value):
self._set_attr('jid', str(value))
[docs]
def get_jid(self):
return JID(self._get_attr('jid'))
[docs]
class PublishOptions(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'publish-options'
plugin_attrib = 'publish_options'
interfaces = {'publish_options'}
is_extension = True
[docs]
def get_publish_options(self):
config = self.xml.find('{jabber:x:data}x')
if config is None:
return None
form = xep_0004.Form(xml=config)
return form
[docs]
def set_publish_options(self, value):
if value is None:
self.del_publish_options()
else:
if isinstance(value, ElementBase):
self.xml.append(value.xml)
else:
self.xml.append(value)
return self
[docs]
def del_publish_options(self):
config = self.xml.find('{jabber:x:data}x')
if config is not None:
self.xml.remove(config)
self.parent().xml.remove(self.xml)
register_stanza_plugin(Iq, Pubsub)
register_stanza_plugin(Pubsub, Affiliations)
register_stanza_plugin(Pubsub, Configure)
register_stanza_plugin(Pubsub, Create)
register_stanza_plugin(Pubsub, Default)
register_stanza_plugin(Pubsub, Items)
register_stanza_plugin(Pubsub, Options)
register_stanza_plugin(Pubsub, Publish)
register_stanza_plugin(Pubsub, PublishOptions)
register_stanza_plugin(Pubsub, Retract)
register_stanza_plugin(Pubsub, Subscribe)
register_stanza_plugin(Pubsub, Subscription)
register_stanza_plugin(Pubsub, Subscriptions)
register_stanza_plugin(Pubsub, Unsubscribe)
register_stanza_plugin(Affiliations, Affiliation, iterable=True)
register_stanza_plugin(Configure, xep_0004.Form)
register_stanza_plugin(Items, Item, iterable=True)
register_stanza_plugin(Publish, Item, iterable=True)
register_stanza_plugin(Retract, Item)
register_stanza_plugin(Subscribe, Options)
register_stanza_plugin(Subscription, SubscribeOptions)
register_stanza_plugin(Subscriptions, Subscription, iterable=True)