summaryrefslogtreecommitdiff
path: root/sleekxmpp
diff options
context:
space:
mode:
Diffstat (limited to 'sleekxmpp')
-rw-r--r--sleekxmpp/plugins/jobs.py44
-rw-r--r--sleekxmpp/plugins/stanza_pubsub.py45
-rw-r--r--sleekxmpp/plugins/xep_0004.py1
-rw-r--r--sleekxmpp/plugins/xep_0030.py405
-rw-r--r--sleekxmpp/plugins/xep_0060.py10
-rw-r--r--sleekxmpp/stanza/error.py2
-rw-r--r--sleekxmpp/xmlstream/scheduler.py87
-rw-r--r--sleekxmpp/xmlstream/stanzabase.py7
-rw-r--r--sleekxmpp/xmlstream/xmlstream.py21
9 files changed, 501 insertions, 121 deletions
diff --git a/sleekxmpp/plugins/jobs.py b/sleekxmpp/plugins/jobs.py
new file mode 100644
index 00000000..bb2e2554
--- /dev/null
+++ b/sleekxmpp/plugins/jobs.py
@@ -0,0 +1,44 @@
+from . import base
+import logging
+from xml.etree import cElementTree as ET
+
+class jobs(base.base_plugin):
+ def plugin_init(self):
+ self.xep = 'pubsubjob'
+ self.description = "Job distribution over Pubsub"
+
+ def post_init(self):
+ pass
+ #TODO add event
+
+ def createJobNode(self, host, jid, node, config=None):
+ pass
+
+ def createJob(self, host, node, jobid=None, payload=None):
+ return self.xmpp.plugin['xep_0060'].setItem(host, node, ((jobid, payload),))
+
+ def claimJob(self, host, node, jobid, ifrom=None):
+ return self._setState(host, node, jobid, ET.Element('{http://andyet.net/protocol/pubsubjob}claimed'))
+
+ def unclaimJob(self, jobid):
+ return self._setState(host, node, jobid, ET.Element('{http://andyet.net/protocol/pubsubjob}unclaimed'))
+
+ def finishJob(self, host, node, jobid, payload=None):
+ finished = ET.Element('{http://andyet.net/protocol/pubsubjob}finished')
+ if payload is not None:
+ finished.append(payload)
+ return self._setState(host, node, jobid, finished)
+
+ def _setState(self, host, node, jobid, state, ifrom=None):
+ iq = self.xmpp.Iq()
+ iq['to'] = host
+ if ifrom: iq['from'] = ifrom
+ iq['type'] = 'set'
+ iq['psstate']['node'] = node
+ iq['psstate']['item'] = jobid
+ iq['psstate']['payload'] = state
+ result = iq.send()
+ if result is None or result['type'] != 'result':
+ return False
+ return True
+
diff --git a/sleekxmpp/plugins/stanza_pubsub.py b/sleekxmpp/plugins/stanza_pubsub.py
index 1dd73d99..1a1526f0 100644
--- a/sleekxmpp/plugins/stanza_pubsub.py
+++ b/sleekxmpp/plugins/stanza_pubsub.py
@@ -10,6 +10,39 @@ def stanzaPlugin(stanza, plugin):
stanza.plugin_attrib_map[plugin.plugin_attrib] = plugin
stanza.plugin_tag_map["{%s}%s" % (plugin.namespace, plugin.name)] = plugin
+class PubsubState(ElementBase):
+ namespace = 'http://jabber.org/protocol/psstate'
+ name = 'state'
+ plugin_attrib = 'psstate'
+ interfaces = set(('node', 'item', 'payload'))
+ plugin_attrib_map = {}
+ plugin_tag_map = {}
+
+ def setPayload(self, value):
+ self.xml.append(value)
+
+ def getPayload(self):
+ childs = self.xml.getchildren()
+ if len(childs) > 0:
+ return childs[0]
+
+ def delPayload(self):
+ for child in self.xml.getchildren():
+ self.xml.remove(child)
+
+stanzaPlugin(Iq, PubsubState)
+
+class PubsubStateEvent(ElementBase):
+ namespace = 'http://jabber.org/protocol/psstate#event'
+ name = 'event'
+ plugin_attrib = 'psstate_event'
+ intefaces = set(tuple())
+ plugin_attrib_map = {}
+ plugin_tag_map = {}
+
+stanzaPlugin(Message, PubsubStateEvent)
+stanzaPlugin(PubsubStateEvent, PubsubState)
+
class Pubsub(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'pubsub'
@@ -321,18 +354,6 @@ class Options(ElementBase):
stanzaPlugin(Pubsub, Options)
stanzaPlugin(Subscribe, Options)
-#iq = Iq()
-#iq['pubsub']['defaultconfig']
-#print(iq)
-
-#from xml.etree import cElementTree as ET
-#iq = Iq()
-#item = Item()
-#item['payload'] = ET.Element("{http://netflint.net/p/crap}stupidshit")
-#item['id'] = 'aa11bbcc'
-#iq['pubsub']['items'].append(item)
-#print(iq)
-
class OwnerAffiliations(Affiliations):
namespace = 'http://jabber.org/protocol/pubsub#owner'
interfaces = set(('node'))
diff --git a/sleekxmpp/plugins/xep_0004.py b/sleekxmpp/plugins/xep_0004.py
index 56d18929..015bd8bc 100644
--- a/sleekxmpp/plugins/xep_0004.py
+++ b/sleekxmpp/plugins/xep_0004.py
@@ -188,7 +188,6 @@ class Form(FieldContainer):
#def getXML(self, tostring = False):
def getXML(self, ftype=None):
- logging.debug("creating form as %s" % ftype)
if ftype:
self.type = ftype
form = ET.Element('{jabber:x:data}x')
diff --git a/sleekxmpp/plugins/xep_0030.py b/sleekxmpp/plugins/xep_0030.py
index 5432dd56..6a31d243 100644
--- a/sleekxmpp/plugins/xep_0030.py
+++ b/sleekxmpp/plugins/xep_0030.py
@@ -1,25 +1,184 @@
"""
- SleekXMPP: The Sleek XMPP Library
- Copyright (C) 2007 Nathanael C. Fritz
- This file is part of SleekXMPP.
-
- SleekXMPP 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 2 of the License, or
- (at your option) any later version.
-
- SleekXMPP 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 SleekXMPP; if not, write to the Free Software
- Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
+ SleekXMPP: The Sleek XMPP Library
+ Copyright (C) 2010 Nathanael C. Fritz, Lance J.T. Stout
+ This file is part of SleekXMPP.
+
+ See the file license.txt for copying permissio
"""
-from . import base
+
import logging
-from xml.etree import cElementTree as ET
+from . import base
+from .. xmlstream.handler.callback import Callback
+from .. xmlstream.matcher.xpath import MatchXPath
+from .. xmlstream.stanzabase import ElementBase, ET, JID
+from .. stanza.iq import Iq
+
+class DiscoInfo(ElementBase):
+ namespace = 'http://jabber.org/protocol/disco#info'
+ name = 'query'
+ plugin_attrib = 'disco_info'
+ interfaces = set(('node', 'features', 'identities'))
+
+ def getFeatures(self):
+ features = []
+ featuresXML = self.xml.findall('{%s}feature' % self.namespace)
+ for feature in featuresXML:
+ features.append(feature.attrib['var'])
+ return features
+
+ def setFeatures(self, features):
+ self.delFeatures()
+ for name in features:
+ self.addFeature(name)
+
+ def delFeatures(self):
+ featuresXML = self.xml.findall('{%s}feature' % self.namespace)
+ for feature in featuresXML:
+ self.xml.remove(feature)
+
+ def addFeature(self, feature):
+ featureXML = ET.Element('{%s}feature' % self.namespace,
+ {'var': feature})
+ self.xml.append(featureXML)
+
+ def delFeature(self, feature):
+ featuresXML = self.xml.findall('{%s}feature' % self.namespace)
+ for featureXML in featuresXML:
+ if featureXML.attrib['var'] == feature:
+ self.xml.remove(featureXML)
+
+ def getIdentities(self):
+ ids = []
+ idsXML = self.xml.findall('{%s}identity' % self.namespace)
+ for idXML in idsXML:
+ idData = (idXML.attrib['category'],
+ idXML.attrib['type'],
+ idXML.attrib.get('name', ''))
+ ids.append(idData)
+ return ids
+
+ def setIdentities(self, ids):
+ self.delIdentities()
+ for idData in ids:
+ self.addIdentity(*idData)
+
+ def delIdentities(self):
+ idsXML = self.xml.findall('{%s}identity' % self.namespace)
+ for idXML in idsXML:
+ self.xml.remove(idXML)
+
+ def addIdentity(self, category, id_type, name=''):
+ idXML = ET.Element('{%s}identity' % self.namespace,
+ {'category': category,
+ 'type': id_type,
+ 'name': name})
+ self.xml.append(idXML)
+
+ def delIdentity(self, category, id_type, name=''):
+ idsXML = self.xml.findall('{%s}identity' % self.namespace)
+ for idXML in idsXML:
+ idData = (idXML.attrib['category'],
+ idXML.attrib['type'])
+ delId = (category, id_type)
+ if idData == delId:
+ self.xml.remove(idXML)
+
+
+class DiscoItems(ElementBase):
+ namespace = 'http://jabber.org/protocol/disco#items'
+ name = 'query'
+ plugin_attrib = 'disco_items'
+ interfaces = set(('node', 'items'))
+
+ def getItems(self):
+ items = []
+ itemsXML = self.xml.findall('{%s}item' % self.namespace)
+ for item in itemsXML:
+ itemData = (item.attrib['jid'],
+ item.attrib.get('node'),
+ item.attrib.get('name'))
+ items.append(itemData)
+ return items
+
+ def setItems(self, items):
+ self.delItems()
+ for item in items:
+ self.addItem(*item)
+
+ def delItems(self):
+ itemsXML = self.xml.findall('{%s}item' % self.namespace)
+ for item in itemsXML:
+ self.xml.remove(item)
+
+ def addItem(self, jid, node='', name=''):
+ itemXML = ET.Element('{%s}item' % self.namespace, {'jid': jid})
+ if name:
+ itemXML.attrib['name'] = name
+ if node:
+ itemXML.attrib['node'] = node
+ self.xml.append(itemXML)
+
+ def delItem(self, jid, node=''):
+ itemsXML = self.xml.findall('{%s}item' % self.namespace)
+ for itemXML in itemsXML:
+ itemData = (itemXML.attrib['jid'],
+ itemXML.attrib.get('node', ''))
+ itemDel = (jid, node)
+ if itemData == itemDel:
+ self.xml.remove(itemXML)
+
+
+class DiscoNode(object):
+ """
+ Collection object for grouping info and item information
+ into nodes.
+ """
+ def __init__(self, name):
+ self.name = name
+ self.info = DiscoInfo()
+ self.items = DiscoItems()
+
+ # This is a bit like poor man's inheritance, but
+ # to simplify adding information to the node we
+ # map node functions to either the info or items
+ # stanza objects.
+ #
+ # We don't want to make DiscoNode inherit from
+ # DiscoInfo and DiscoItems because DiscoNode is
+ # not an actual stanza, and doing so would create
+ # confusion and potential bugs.
+
+ self._map(self.items, 'items', ['get', 'set', 'del'])
+ self._map(self.items, 'item', ['add', 'del'])
+ self._map(self.info, 'identities', ['get', 'set', 'del'])
+ self._map(self.info, 'identity', ['add', 'del'])
+ self._map(self.info, 'features', ['get', 'set', 'del'])
+ self._map(self.info, 'feature', ['add', 'del'])
+
+ def isEmpty(self):
+ """
+ Test if the node contains any information. Useful for
+ determining if a node can be deleted.
+ """
+ ids = self.getIdentities()
+ features = self.getFeatures()
+ items = self.getItems()
+
+ if not ids and not features and not items:
+ return True
+ return False
+
+ def _map(self, obj, interface, access):
+ """
+ Map functions of the form obj.accessInterface
+ to self.accessInterface for each given access type.
+ """
+ interface = interface.title()
+ for access_type in access:
+ method = access_type + interface
+ if hasattr(obj, method):
+ setattr(self, method, getattr(obj, method))
+
class xep_0030(base.base_plugin):
"""
@@ -29,85 +188,137 @@ class xep_0030(base.base_plugin):
def plugin_init(self):
self.xep = '0030'
self.description = 'Service Discovery'
- self.features = {'main': ['http://jabber.org/protocol/disco#info', 'http://jabber.org/protocol/disco#items']}
- self.identities = {'main': [{'category': 'client', 'type': 'pc', 'name': 'SleekXMPP'}]}
- self.items = {'main': []}
- self.xmpp.add_handler("<iq type='get' xmlns='%s'><query xmlns='http://jabber.org/protocol/disco#info' /></iq>" % self.xmpp.default_ns, self.info_handler)
- self.xmpp.add_handler("<iq type='get' xmlns='%s'><query xmlns='http://jabber.org/protocol/disco#items' /></iq>" % self.xmpp.default_ns, self.item_handler)
+
+ self.xmpp.registerHandler(
+ Callback('Disco Items',
+ MatchXPath('{%s}iq/{%s}query' % (self.xmpp.default_ns,
+ DiscoItems.namespace)),
+ self.handle_item_query))
+
+ self.xmpp.registerHandler(
+ Callback('Disco Info',
+ MatchXPath('{%s}iq/{%s}query' % (self.xmpp.default_ns,
+ DiscoInfo.namespace)),
+ self.handle_info_query))
+
+ self.xmpp.stanzaPlugin(Iq, DiscoInfo)
+ self.xmpp.stanzaPlugin(Iq, DiscoItems)
+
+ self.xmpp.add_event_handler('disco_items_request', self.handle_disco_items)
+ self.xmpp.add_event_handler('disco_info_request', self.handle_disco_info)
+
+ self.nodes = {'main': DiscoNode('main')}
+
+ def add_node(self, node):
+ if node not in self.nodes:
+ self.nodes[node] = DiscoNode(node)
+
+ def del_node(self, node):
+ if node in self.nodes:
+ del self.nodes[node]
+
+ def handle_item_query(self, iq):
+ if iq['type'] == 'get':
+ logging.debug("Items requested by %s" % iq['from'])
+ self.xmpp.event('disco_items_request', iq)
+ elif iq['type'] == 'result':
+ logging.debug("Items result from %s" % iq['from'])
+ self.xmpp.event('disco_items', iq)
+
+ def handle_info_query(self, iq):
+ if iq['type'] == 'get':
+ logging.debug("Info requested by %s" % iq['from'])
+ self.xmpp.event('disco_info_request', iq)
+ elif iq['type'] == 'result':
+ logging.debug("Info result from %s" % iq['from'])
+ self.xmpp.event('disco_info', iq)
+
+ def handle_disco_info(self, iq, forwarded=False):
+ """
+ A default handler for disco#info requests. If another
+ handler is registered, this one will defer and not run.
+ """
+ handlers = self.xmpp.event_handlers['disco_info_request']
+ if not forwarded and len(handlers) > 1:
+ return
+
+ node_name = iq['disco_info']['node']
+ if not node_name:
+ node_name = 'main'
+
+ logging.debug("Using default handler for disco#info on node '%s'." % node_name)
+
+ if node_name in self.nodes:
+ node = self.nodes[node_name]
+ iq.reply().setPayload(node.info.xml).send()
+ else:
+ logging.debug("Node %s requested, but does not exist." % node_name)
+ iq.reply().error().setPayload(iq['disco_info'].xml)
+ iq['error']['code'] = '404'
+ iq['error']['type'] = 'cancel'
+ iq['error']['condition'] = 'item-not-found'
+ iq.send()
+
+ def handle_disco_items(self, iq, forwarded=False):
+ """
+ A default handler for disco#items requests. If another
+ handler is registered, this one will defer and not run.
+
+ If this handler is called by your own custom handler with
+ forwarded set to True, then it will run as normal.
+ """
+ handlers = self.xmpp.event_handlers['disco_items_request']
+ if not forwarded and len(handlers) > 1:
+ return
+
+ node_name = iq['disco_items']['node']
+ if not node_name:
+ node_name = 'main'
+
+ logging.debug("Using default handler for disco#items on node '%s'." % node_name)
+
+ if node_name in self.nodes:
+ node = self.nodes[node_name]
+ iq.reply().setPayload(node.items.xml).send()
+ else:
+ logging.debug("Node %s requested, but does not exist." % node_name)
+ iq.reply().error().setPayload(iq['disco_items'].xml)
+ iq['error']['code'] = '404'
+ iq['error']['type'] = 'cancel'
+ iq['error']['condition'] = 'item-not-found'
+ iq.send()
+
+ # Older interface methods for backwards compatibility
+
+ def getInfo(self, jid, node=''):
+ iq = self.xmpp.Iq()
+ iq['type'] = 'get'
+ iq['to'] = jid
+ iq['from'] = self.xmpp.fulljid
+ iq['disco_info']['node'] = node
+ iq.send()
+
+ def getItems(self, jid, node=''):
+ iq = self.xmpp.Iq()
+ iq['type'] = 'get'
+ iq['to'] = jid
+ iq['from'] = self.xmpp.fulljid
+ iq['disco_items']['node'] = node
+ iq.send()
def add_feature(self, feature, node='main'):
- if not node in self.features:
- self.features[node] = []
- self.features[node].append(feature)
-
- def add_identity(self, category=None, itype=None, name=None, node='main'):
- if not node in self.identities:
- self.identities[node] = []
- self.identities[node].append({'category': category, 'type': itype, 'name': name})
+ self.add_node(node)
+ self.nodes[node].addFeature(feature)
- def add_item(self, jid=None, name=None, node='main', subnode=''):
- if not node in self.items:
- self.items[node] = []
- self.items[node].append({'jid': jid, 'name': name, 'node': subnode})
-
- def info_handler(self, xml):
- logging.debug("Info request from %s" % xml.get('from', ''))
- iq = self.xmpp.makeIqResult(xml.get('id', self.xmpp.getNewId()))
- iq.attrib['from'] = xml.get('to')
- iq.attrib['to'] = xml.get('from', self.xmpp.server)
- query = xml.find('{http://jabber.org/protocol/disco#info}query')
- node = query.get('node', 'main')
- for identity in self.identities.get(node, []):
- idxml = ET.Element('identity')
- for attrib in identity:
- if identity[attrib]:
- idxml.attrib[attrib] = identity[attrib]
- query.append(idxml)
- for feature in self.features.get(node, []):
- featxml = ET.Element('feature')
- featxml.attrib['var'] = feature
- query.append(featxml)
- iq.append(query)
- #print ET.tostring(iq)
- self.xmpp.send(iq)
-
- def item_handler(self, xml):
- logging.debug("Item request from %s" % xml.get('from', ''))
- iq = self.xmpp.makeIqResult(xml.get('id', self.xmpp.getNewId()))
- iq.attrib['from'] = xml.get('to')
- iq.attrib['to'] = xml.get('from', self.xmpp.server)
- query = self.xmpp.makeIqQuery(iq, 'http://jabber.org/protocol/disco#items').find('{http://jabber.org/protocol/disco#items}query')
- node = xml.find('{http://jabber.org/protocol/disco#items}query').get('node', 'main')
- for item in self.items.get(node, []):
- itemxml = ET.Element('item')
- itemxml.attrib = item
- if itemxml.attrib['jid'] is None:
- itemxml.attrib['jid'] = xml.get('to')
- query.append(itemxml)
- self.xmpp.send(iq)
+ def add_identity(self, category='', itype='', name='', node='main'):
+ self.add_node(node)
+ self.nodes[node].addIdentity(category=category,
+ id_type=itype,
+ name=name)
- def getItems(self, jid, node=None):
- iq = self.xmpp.makeIqGet()
- iq.attrib['from'] = self.xmpp.fulljid
- iq.attrib['to'] = jid
- self.xmpp.makeIqQuery(iq, 'http://jabber.org/protocol/disco#items')
- if node:
- iq.find('{http://jabber.org/protocol/disco#items}query').attrib['node'] = node
- return iq.send()
-
- def getInfo(self, jid, node=None):
- iq = self.xmpp.makeIqGet()
- iq.attrib['from'] = self.xmpp.fulljid
- iq.attrib['to'] = jid
- self.xmpp.makeIqQuery(iq, 'http://jabber.org/protocol/disco#info')
- if node:
- iq.find('{http://jabber.org/protocol/disco#info}query').attrib['node'] = node
- return iq.send()
-
- def parseInfo(self, xml):
- result = {'identity': {}, 'feature': []}
- for identity in xml.findall('{http://jabber.org/protocol/disco#info}query/{{http://jabber.org/protocol/disco#info}identity'):
- result['identity'][identity['name']] = identity.attrib
- for feature in xml.findall('{http://jabber.org/protocol/disco#info}query/{{http://jabber.org/protocol/disco#info}feature'):
- result['feature'].append(feature.get('var', '__unknown__'))
- return result
+ def add_item(self, jid=None, name='', node='main', subnode=''):
+ self.add_node(node)
+ self.add_node(subnode)
+ if jid is None:
+ jid = self.xmpp.fulljid
+ self.nodes[node].addItem(jid=jid, name=name, node=subnode)
diff --git a/sleekxmpp/plugins/xep_0060.py b/sleekxmpp/plugins/xep_0060.py
index 44a70e9a..bff158a0 100644
--- a/sleekxmpp/plugins/xep_0060.py
+++ b/sleekxmpp/plugins/xep_0060.py
@@ -14,12 +14,14 @@ class xep_0060(base.base_plugin):
self.xep = '0060'
self.description = 'Publish-Subscribe'
- def create_node(self, jid, node, config=None, collection=False):
+ def create_node(self, jid, node, config=None, collection=False, ntype=None):
pubsub = ET.Element('{http://jabber.org/protocol/pubsub}pubsub')
create = ET.Element('create')
create.set('node', node)
pubsub.append(create)
configure = ET.Element('configure')
+ if collection:
+ ntype = 'collection'
#if config is None:
# submitform = self.xmpp.plugin['xep_0004'].makeForm('submit')
#else:
@@ -29,11 +31,11 @@ class xep_0060(base.base_plugin):
submitform.field['FORM_TYPE'].setValue('http://jabber.org/protocol/pubsub#node_config')
else:
submitform.addField('FORM_TYPE', 'hidden', value='http://jabber.org/protocol/pubsub#node_config')
- if collection:
+ if ntype:
if 'pubsub#node_type' in submitform.field:
- submitform.field['pubsub#node_type'].setValue('collection')
+ submitform.field['pubsub#node_type'].setValue(ntype)
else:
- submitform.addField('pubsub#node_type', value='collection')
+ submitform.addField('pubsub#node_type', value=ntype)
else:
if 'pubsub#node_type' in submitform.field:
submitform.field['pubsub#node_type'].setValue('leaf')
diff --git a/sleekxmpp/stanza/error.py b/sleekxmpp/stanza/error.py
index f87b6490..ee46722a 100644
--- a/sleekxmpp/stanza/error.py
+++ b/sleekxmpp/stanza/error.py
@@ -11,7 +11,7 @@ class Error(ElementBase):
namespace = 'jabber:client'
name = 'error'
plugin_attrib = 'error'
- conditions = set(('bad-request', 'conflict', 'feature-not-implemented', 'forbidden', 'gone', 'item-not-found', 'jid-malformed', 'not-acceptable', 'not-allowed', 'not-authorized', 'payment-required', 'recipient-unavailable', 'redirect', 'registration-required', 'remote-server-not-found', 'remote-server-timeout', 'service-unavailable', 'subscription-required', 'undefined-condition', 'unexpected-request'))
+ conditions = set(('bad-request', 'conflict', 'feature-not-implemented', 'forbidden', 'gone', 'internal-server-error', 'item-not-found', 'jid-malformed', 'not-acceptable', 'not-allowed', 'not-authorized', 'payment-required', 'recipient-unavailable', 'redirect', 'registration-required', 'remote-server-not-found', 'remote-server-timeout', 'resource-constraint', 'service-unavailable', 'subscription-required', 'undefined-condition', 'unexpected-request'))
interfaces = set(('code', 'condition', 'text', 'type'))
types = set(('cancel', 'continue', 'modify', 'auth', 'wait'))
sub_interfaces = set(('text',))
diff --git a/sleekxmpp/xmlstream/scheduler.py b/sleekxmpp/xmlstream/scheduler.py
new file mode 100644
index 00000000..945d9fad
--- /dev/null
+++ b/sleekxmpp/xmlstream/scheduler.py
@@ -0,0 +1,87 @@
+try:
+ import queue
+except ImportError:
+ import Queue as queue
+import time
+import threading
+import logging
+
+class Task(object):
+ """Task object for the Scheduler class"""
+ def __init__(self, name, seconds, callback, args=None, kwargs=None, repeat=False, qpointer=None):
+ self.name = name
+ self.seconds = seconds
+ self.callback = callback
+ self.args = args or tuple()
+ self.kwargs = kwargs or {}
+ self.repeat = repeat
+ self.next = time.time() + self.seconds
+ self.qpointer = qpointer
+
+ def run(self):
+ if self.qpointer is not None:
+ self.qpointer.put(('schedule', self.callback, self.args))
+ else:
+ self.callback(*self.args, **self.kwargs)
+ self.reset()
+ return self.repeat
+
+ def reset(self):
+ self.next = time.time() + self.seconds
+
+class Scheduler(object):
+ """Threaded scheduler that allows for updates mid-execution unlike http://docs.python.org/library/sched.html#module-sched"""
+ def __init__(self, parentqueue=None):
+ self.addq = queue.Queue()
+ self.schedule = []
+ self.thread = None
+ self.run = False
+ self.parentqueue = parentqueue
+
+ def process(self, threaded=True):
+ if threaded:
+ self.thread = threading.Thread(name='shedulerprocess', target=self._process)
+ self.thread.start()
+ else:
+ self._process()
+
+ def _process(self):
+ self.run = True
+ while self.run:
+ try:
+ wait = 1
+ updated = False
+ if self.schedule:
+ wait = self.schedule[0].next - time.time()
+ try:
+ if wait <= 0.0:
+ newtask = self.addq.get(False)
+ else:
+ newtask = self.addq.get(True, wait)
+ except queue.Empty:
+ cleanup = []
+ for task in self.schedule:
+ if time.time() >= task.next:
+ updated = True
+ if not task.run():
+ cleanup.append(task)
+ else:
+ break
+ for task in cleanup:
+ x = self.schedule.pop(self.schedule.index(task))
+ else:
+ updated = True
+ self.schedule.append(newtask)
+ finally:
+ if updated: self.schedule = sorted(self.schedule, key=lambda task: task.next)
+ except KeyboardInterrupt:
+ self.run = False
+ logging.debug("Qutting Scheduler thread")
+ if self.parentqueue is not None:
+ self.parentqueue.put(('quit', None, None))
+
+ def add(self, name, seconds, callback, args=None, kwargs=None, repeat=False, qpointer=None):
+ self.addq.put(Task(name, seconds, callback, args, kwargs, repeat, qpointer))
+
+ def quit(self):
+ self.run = False
diff --git a/sleekxmpp/xmlstream/stanzabase.py b/sleekxmpp/xmlstream/stanzabase.py
index 3f3f5e08..64020c8f 100644
--- a/sleekxmpp/xmlstream/stanzabase.py
+++ b/sleekxmpp/xmlstream/stanzabase.py
@@ -78,6 +78,9 @@ class ElementBase(tostring.ToString):
def __iter__(self):
self.idx = 0
return self
+
+ def __bool__(self):
+ return True
def __next__(self):
self.idx += 1
@@ -319,6 +322,8 @@ class StanzaBase(ElementBase):
def __init__(self, stream=None, xml=None, stype=None, sto=None, sfrom=None, sid=None):
self.stream = stream
+ if stream is not None:
+ self.namespace = stream.default_ns
ElementBase.__init__(self, xml)
if stype is not None:
self['type'] = stype
@@ -326,8 +331,6 @@ class StanzaBase(ElementBase):
self['to'] = sto
if sfrom is not None:
self['from'] = sfrom
- if stream is not None:
- self.namespace = stream.default_ns
self.tag = "{%s}%s" % (self.namespace, self.name)
def setType(self, value):
diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py
index 025884b7..6b92abca 100644
--- a/sleekxmpp/xmlstream/xmlstream.py
+++ b/sleekxmpp/xmlstream/xmlstream.py
@@ -22,6 +22,7 @@ import time
import traceback
import types
import xml.sax.saxutils
+from . import scheduler
HANDLER_THREADS = 1
@@ -75,6 +76,7 @@ class XMLStream(object):
self.eventqueue = queue.Queue()
self.sendqueue = queue.Queue()
+ self.scheduler = scheduler.Scheduler(self.eventqueue)
self.namespace_map = {}
@@ -145,7 +147,9 @@ class XMLStream(object):
raise RestartStream()
def process(self, threaded=True):
+ self.scheduler.process(threaded=True)
for t in range(0, HANDLER_THREADS):
+ logging.debug("Starting HANDLER THREAD")
self.__thread['eventhandle%s' % t] = threading.Thread(name='eventhandle%s' % t, target=self._eventRunner)
self.__thread['eventhandle%s' % t].start()
self.__thread['sendthread'] = threading.Thread(name='sendthread', target=self._sendThread)
@@ -156,8 +160,8 @@ class XMLStream(object):
else:
self._process()
- def schedule(self, seconds, handler, args=None):
- threading.Timer(seconds, handler, args).start()
+ def schedule(self, name, seconds, callback, args=None, kwargs=None, repeat=False):
+ self.scheduler.add(name, seconds, callback, args, kwargs, repeat, qpointer=self.eventqueue)
def _process(self):
"Start processing the socket."
@@ -177,6 +181,7 @@ class XMLStream(object):
self.state.set('reconnect', False)
self.disconnect()
self.run = False
+ self.scheduler.run = False
self.eventqueue.put(('quit', None, None))
return
except CloseStream:
@@ -223,6 +228,7 @@ class XMLStream(object):
edepth += -1
if edepth == 0 and event == b'end':
self.disconnect(reconnect=self.state['reconnect'])
+ logging.debug("Ending readXML loop")
return False
elif edepth == 1:
#self.xmlin.put(xmlobj)
@@ -231,11 +237,13 @@ class XMLStream(object):
except RestartStream:
return True
except CloseStream:
+ logging.debug("Ending readXML loop")
return False
if root:
root.clear()
if event == b'start':
edepth += 1
+ logging.debug("Ending readXML loop")
def _sendThread(self):
while self.run:
@@ -265,6 +273,7 @@ class XMLStream(object):
logging.debug("Disconnecting...")
self.state.set('disconnecting', True)
self.run = False
+ self.scheduler.run = False
if self.state['connected']:
self.sendRaw(self.stream_footer)
time.sleep(1)
@@ -323,6 +332,9 @@ class XMLStream(object):
event = self.eventqueue.get(True, timeout=5)
except queue.Empty:
event = None
+ except KeyboardInterrupt:
+ self.run = False
+ self.scheduler.run = False
if event is not None:
etype = event[0]
handler = event[1]
@@ -334,9 +346,10 @@ class XMLStream(object):
except Exception as e:
traceback.print_exc()
args[0].exception(e)
- elif etype == 'sched':
+ elif etype == 'schedule':
try:
- handler.run(*args)
+ logging.debug(args)
+ handler(*args[0])
except:
logging.error(traceback.format_exc())
elif etype == 'quit':
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-03 15:30:02 +0000